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
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(self, collation: str) -> CollationClause: ...
1118 def between(
1119 self, cleft: Any, cright: Any, symmetric: bool = False
1120 ) -> BinaryExpression[bool]: ...
1122 def distinct(self: _SQO[_T_co]) -> UnaryExpression[_T_co]: ...
1124 def any_(self) -> CollectionAggregate[Any]: ...
1126 def all_(self) -> CollectionAggregate[Any]: ...
1128 # numeric overloads. These need more tweaking
1129 # in particular they all need to have a variant for Optional[_T]
1130 # because Optional only applies to the data side, not the expression
1131 # side
1133 @overload
1134 def __add__(
1135 self: _SQO[_NMT],
1136 other: Any,
1137 ) -> ColumnElement[_NMT]: ...
1139 @overload
1140 def __add__(
1141 self: _SQO[str],
1142 other: Any,
1143 ) -> ColumnElement[str]: ...
1145 @overload
1146 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1148 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1150 @overload
1151 def __radd__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1153 @overload
1154 def __radd__(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
1156 def __radd__(self, other: Any) -> ColumnElement[Any]: ...
1158 @overload
1159 def __sub__(
1160 self: _SQO[_NMT],
1161 other: Any,
1162 ) -> ColumnElement[_NMT]: ...
1164 @overload
1165 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1167 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1169 @overload
1170 def __rsub__(
1171 self: _SQO[_NMT],
1172 other: Any,
1173 ) -> ColumnElement[_NMT]: ...
1175 @overload
1176 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1178 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1180 @overload
1181 def __mul__(
1182 self: _SQO[_NMT],
1183 other: Any,
1184 ) -> ColumnElement[_NMT]: ...
1186 @overload
1187 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1189 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1191 @overload
1192 def __rmul__(
1193 self: _SQO[_NMT],
1194 other: Any,
1195 ) -> ColumnElement[_NMT]: ...
1197 @overload
1198 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1200 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1202 @overload
1203 def __mod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1205 @overload
1206 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1208 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1210 @overload
1211 def __rmod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1213 @overload
1214 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1216 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1218 @overload
1219 def __truediv__(
1220 self: _SQO[int], other: Any
1221 ) -> ColumnElement[_NUMERIC]: ...
1223 @overload
1224 def __truediv__(self: _SQO[_NT], other: Any) -> ColumnElement[_NT]: ...
1226 @overload
1227 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1229 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1231 @overload
1232 def __rtruediv__(
1233 self: _SQO[_NMT], other: Any
1234 ) -> ColumnElement[_NUMERIC]: ...
1236 @overload
1237 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1239 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1241 @overload
1242 def __floordiv__(
1243 self: _SQO[_NMT], other: Any
1244 ) -> ColumnElement[_NMT]: ...
1246 @overload
1247 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1249 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1251 @overload
1252 def __rfloordiv__(
1253 self: _SQO[_NMT], other: Any
1254 ) -> ColumnElement[_NMT]: ...
1256 @overload
1257 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1259 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1262class SQLColumnExpression(
1263 SQLCoreOperations[_T_co], roles.ExpressionElementRole[_T_co], TypingOnly
1264):
1265 """A type that may be used to indicate any SQL column element or object
1266 that acts in place of one.
1268 :class:`.SQLColumnExpression` is a base of
1269 :class:`.ColumnElement`, as well as within the bases of ORM elements
1270 such as :class:`.InstrumentedAttribute`, and may be used in :pep:`484`
1271 typing to indicate arguments or return values that should behave
1272 as column expressions.
1274 .. versionadded:: 2.0.0b4
1277 """
1279 __slots__ = ()
1282_SQO = SQLCoreOperations
1285class ColumnElement(
1286 roles.ColumnArgumentOrKeyRole,
1287 roles.StatementOptionRole,
1288 roles.WhereHavingRole,
1289 roles.BinaryElementRole[_T],
1290 roles.OrderByRole,
1291 roles.ColumnsClauseRole,
1292 roles.LimitOffsetRole,
1293 roles.DMLColumnRole,
1294 roles.DDLConstraintColumnRole,
1295 roles.DDLExpressionRole,
1296 SQLColumnExpression[_T],
1297 DQLDMLClauseElement,
1298):
1299 """Represent a column-oriented SQL expression suitable for usage in the
1300 "columns" clause, WHERE clause etc. of a statement.
1302 While the most familiar kind of :class:`_expression.ColumnElement` is the
1303 :class:`_schema.Column` object, :class:`_expression.ColumnElement`
1304 serves as the basis
1305 for any unit that may be present in a SQL expression, including
1306 the expressions themselves, SQL functions, bound parameters,
1307 literal expressions, keywords such as ``NULL``, etc.
1308 :class:`_expression.ColumnElement`
1309 is the ultimate base class for all such elements.
1311 A wide variety of SQLAlchemy Core functions work at the SQL expression
1312 level, and are intended to accept instances of
1313 :class:`_expression.ColumnElement` as
1314 arguments. These functions will typically document that they accept a
1315 "SQL expression" as an argument. What this means in terms of SQLAlchemy
1316 usually refers to an input which is either already in the form of a
1317 :class:`_expression.ColumnElement` object,
1318 or a value which can be **coerced** into
1319 one. The coercion rules followed by most, but not all, SQLAlchemy Core
1320 functions with regards to SQL expressions are as follows:
1322 * a literal Python value, such as a string, integer or floating
1323 point value, boolean, datetime, ``Decimal`` object, or virtually
1324 any other Python object, will be coerced into a "literal bound
1325 value". This generally means that a :func:`.bindparam` will be
1326 produced featuring the given value embedded into the construct; the
1327 resulting :class:`.BindParameter` object is an instance of
1328 :class:`_expression.ColumnElement`.
1329 The Python value will ultimately be sent
1330 to the DBAPI at execution time as a parameterized argument to the
1331 ``execute()`` or ``executemany()`` methods, after SQLAlchemy
1332 type-specific converters (e.g. those provided by any associated
1333 :class:`.TypeEngine` objects) are applied to the value.
1335 * any special object value, typically ORM-level constructs, which
1336 feature an accessor called ``__clause_element__()``. The Core
1337 expression system looks for this method when an object of otherwise
1338 unknown type is passed to a function that is looking to coerce the
1339 argument into a :class:`_expression.ColumnElement` and sometimes a
1340 :class:`_expression.SelectBase` expression.
1341 It is used within the ORM to
1342 convert from ORM-specific objects like mapped classes and
1343 mapped attributes into Core expression objects.
1345 * The Python ``None`` value is typically interpreted as ``NULL``,
1346 which in SQLAlchemy Core produces an instance of :func:`.null`.
1348 A :class:`_expression.ColumnElement` provides the ability to generate new
1349 :class:`_expression.ColumnElement`
1350 objects using Python expressions. This means that Python operators
1351 such as ``==``, ``!=`` and ``<`` are overloaded to mimic SQL operations,
1352 and allow the instantiation of further :class:`_expression.ColumnElement`
1353 instances
1354 which are composed from other, more fundamental
1355 :class:`_expression.ColumnElement`
1356 objects. For example, two :class:`.ColumnClause` objects can be added
1357 together with the addition operator ``+`` to produce
1358 a :class:`.BinaryExpression`.
1359 Both :class:`.ColumnClause` and :class:`.BinaryExpression` are subclasses
1360 of :class:`_expression.ColumnElement`:
1362 .. sourcecode:: pycon+sql
1364 >>> from sqlalchemy.sql import column
1365 >>> column("a") + column("b")
1366 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
1367 >>> print(column("a") + column("b"))
1368 {printsql}a + b
1370 .. seealso::
1372 :class:`_schema.Column`
1374 :func:`_expression.column`
1376 """
1378 __visit_name__ = "column_element"
1380 primary_key: bool = False
1381 _is_clone_of: Optional[ColumnElement[_T]]
1382 _is_column_element = True
1383 _insert_sentinel: bool = False
1384 _omit_from_statements = False
1385 _is_collection_aggregate = False
1387 foreign_keys: AbstractSet[ForeignKey] = frozenset()
1389 @util.memoized_property
1390 def _proxies(self) -> List[ColumnElement[Any]]:
1391 return []
1393 @util.non_memoized_property
1394 def _tq_label(self) -> Optional[str]:
1395 """The named label that can be used to target
1396 this column in a result set in a "table qualified" context.
1398 This label is almost always the label used when
1399 rendering <expr> AS <label> in a SELECT statement when using
1400 the LABEL_STYLE_TABLENAME_PLUS_COL label style, which is what the
1401 legacy ORM ``Query`` object uses as well.
1403 For a regular Column bound to a Table, this is typically the label
1404 <tablename>_<columnname>. For other constructs, different rules
1405 may apply, such as anonymized labels and others.
1407 .. versionchanged:: 1.4.21 renamed from ``._label``
1409 """
1410 return None
1412 key: Optional[str] = None
1413 """The 'key' that in some circumstances refers to this object in a
1414 Python namespace.
1416 This typically refers to the "key" of the column as present in the
1417 ``.c`` collection of a selectable, e.g. ``sometable.c["somekey"]`` would
1418 return a :class:`_schema.Column` with a ``.key`` of "somekey".
1420 """
1422 @HasMemoized.memoized_attribute
1423 def _tq_key_label(self) -> Optional[str]:
1424 """A label-based version of 'key' that in some circumstances refers
1425 to this object in a Python namespace.
1428 _tq_key_label comes into play when a select() statement is constructed
1429 with apply_labels(); in this case, all Column objects in the ``.c``
1430 collection are rendered as <tablename>_<columnname> in SQL; this is
1431 essentially the value of ._label. But to locate those columns in the
1432 ``.c`` collection, the name is along the lines of <tablename>_<key>;
1433 that's the typical value of .key_label.
1435 .. versionchanged:: 1.4.21 renamed from ``._key_label``
1437 """
1438 return self._proxy_key
1440 @property
1441 def _key_label(self) -> Optional[str]:
1442 """legacy; renamed to _tq_key_label"""
1443 return self._tq_key_label
1445 @property
1446 def _label(self) -> Optional[str]:
1447 """legacy; renamed to _tq_label"""
1448 return self._tq_label
1450 @property
1451 def _non_anon_label(self) -> Optional[str]:
1452 """the 'name' that naturally applies this element when rendered in
1453 SQL.
1455 Concretely, this is the "name" of a column or a label in a
1456 SELECT statement; ``<columnname>`` and ``<labelname>`` below:
1458 .. sourcecode:: sql
1460 SELECT <columnmame> FROM table
1462 SELECT column AS <labelname> FROM table
1464 Above, the two names noted will be what's present in the DBAPI
1465 ``cursor.description`` as the names.
1467 If this attribute returns ``None``, it means that the SQL element as
1468 written does not have a 100% fully predictable "name" that would appear
1469 in the ``cursor.description``. Examples include SQL functions, CAST
1470 functions, etc. While such things do return names in
1471 ``cursor.description``, they are only predictable on a
1472 database-specific basis; e.g. an expression like ``MAX(table.col)`` may
1473 appear as the string ``max`` on one database (like PostgreSQL) or may
1474 appear as the whole expression ``max(table.col)`` on SQLite.
1476 The default implementation looks for a ``.name`` attribute on the
1477 object, as has been the precedent established in SQLAlchemy for many
1478 years. An exception is made on the ``FunctionElement`` subclass
1479 so that the return value is always ``None``.
1481 .. versionadded:: 1.4.21
1485 """
1486 return getattr(self, "name", None)
1488 _render_label_in_columns_clause = True
1489 """A flag used by select._columns_plus_names that helps to determine
1490 we are actually going to render in terms of "SELECT <col> AS <label>".
1491 This flag can be returned as False for some Column objects that want
1492 to be rendered as simple "SELECT <col>"; typically columns that don't have
1493 any parent table and are named the same as what the label would be
1494 in any case.
1496 """
1498 _allow_label_resolve = True
1499 """A flag that can be flipped to prevent a column from being resolvable
1500 by string label name.
1502 The joined eager loader strategy in the ORM uses this, for example.
1504 """
1506 _is_implicitly_boolean = False
1508 _alt_names: Sequence[str] = ()
1510 if TYPE_CHECKING:
1512 def _ungroup(self) -> ColumnElement[_T]: ...
1514 @overload
1515 def self_group(self, against: None = None) -> ColumnElement[_T]: ...
1517 @overload
1518 def self_group(
1519 self, against: Optional[OperatorType] = None
1520 ) -> ColumnElement[Any]: ...
1522 def self_group(
1523 self, against: Optional[OperatorType] = None
1524 ) -> ColumnElement[Any]:
1525 if (
1526 against in (operators.and_, operators.or_, operators._asbool)
1527 and self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity
1528 ):
1529 return AsBoolean(self, operators.is_true, operators.is_false)
1530 elif against in (operators.any_op, operators.all_op):
1531 return Grouping(self)
1532 else:
1533 return self
1535 @overload
1536 def _negate(self: ColumnElement[bool]) -> ColumnElement[bool]: ...
1538 @overload
1539 def _negate(self: ColumnElement[_T]) -> ColumnElement[_T]: ...
1541 def _negate(self) -> ColumnElement[Any]:
1542 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
1543 return AsBoolean(self, operators.is_false, operators.is_true)
1544 else:
1545 grouped = self.self_group(against=operators.inv)
1546 assert isinstance(grouped, ColumnElement)
1547 return UnaryExpression(
1548 grouped,
1549 operator=operators.inv,
1550 )
1552 type: TypeEngine[_T]
1554 if not TYPE_CHECKING:
1556 @util.memoized_property
1557 def type(self) -> TypeEngine[_T]: # noqa: A001
1558 # used for delayed setup of
1559 # type_api
1560 return type_api.NULLTYPE
1562 @HasMemoized.memoized_attribute
1563 def comparator(self) -> TypeEngine.Comparator[_T]:
1564 try:
1565 comparator_factory = self.type.comparator_factory
1566 except AttributeError as err:
1567 raise TypeError(
1568 "Object %r associated with '.type' attribute "
1569 "is not a TypeEngine class or object" % self.type
1570 ) from err
1571 else:
1572 return comparator_factory(self)
1574 def __setstate__(self, state):
1575 self.__dict__.update(state)
1577 def __getattr__(self, key: str) -> Any:
1578 try:
1579 return getattr(self.comparator, key)
1580 except AttributeError as err:
1581 raise AttributeError(
1582 "Neither %r object nor %r object has an attribute %r"
1583 % (
1584 type(self).__name__,
1585 type(self.comparator).__name__,
1586 key,
1587 )
1588 ) from err
1590 def operate(
1591 self,
1592 op: operators.OperatorType,
1593 *other: Any,
1594 **kwargs: Any,
1595 ) -> ColumnElement[Any]:
1596 return op(self.comparator, *other, **kwargs) # type: ignore[no-any-return] # noqa: E501
1598 def reverse_operate(
1599 self, op: operators.OperatorType, other: Any, **kwargs: Any
1600 ) -> ColumnElement[Any]:
1601 return op(other, self.comparator, **kwargs) # type: ignore[no-any-return] # noqa: E501
1603 def _bind_param(
1604 self,
1605 operator: operators.OperatorType,
1606 obj: Any,
1607 type_: Optional[TypeEngine[_T]] = None,
1608 expanding: bool = False,
1609 ) -> BindParameter[_T]:
1610 return BindParameter(
1611 None,
1612 obj,
1613 _compared_to_operator=operator,
1614 type_=type_,
1615 _compared_to_type=self.type,
1616 unique=True,
1617 expanding=expanding,
1618 )
1620 @property
1621 def expression(self) -> ColumnElement[Any]:
1622 """Return a column expression.
1624 Part of the inspection interface; returns self.
1626 """
1627 return self
1629 @property
1630 def _select_iterable(self) -> _SelectIterable:
1631 return (self,)
1633 @util.memoized_property
1634 def base_columns(self) -> FrozenSet[ColumnElement[Any]]:
1635 return frozenset(c for c in self.proxy_set if not c._proxies)
1637 @util.memoized_property
1638 def proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1639 """set of all columns we are proxying
1641 as of 2.0 this is explicitly deannotated columns. previously it was
1642 effectively deannotated columns but wasn't enforced. annotated
1643 columns should basically not go into sets if at all possible because
1644 their hashing behavior is very non-performant.
1646 """
1647 return frozenset([self._deannotate()]).union(
1648 itertools.chain(*[c.proxy_set for c in self._proxies])
1649 )
1651 @util.memoized_property
1652 def _expanded_proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1653 return frozenset(_expand_cloned(self.proxy_set))
1655 def _uncached_proxy_list(self) -> List[ColumnElement[Any]]:
1656 """An 'uncached' version of proxy set.
1658 This list includes annotated columns which perform very poorly in
1659 set operations.
1661 """
1663 return [self] + list(
1664 itertools.chain(*[c._uncached_proxy_list() for c in self._proxies])
1665 )
1667 def shares_lineage(self, othercolumn: ColumnElement[Any]) -> bool:
1668 """Return True if the given :class:`_expression.ColumnElement`
1669 has a common ancestor to this :class:`_expression.ColumnElement`."""
1671 return bool(self.proxy_set.intersection(othercolumn.proxy_set))
1673 def _compare_name_for_result(self, other: ColumnElement[Any]) -> bool:
1674 """Return True if the given column element compares to this one
1675 when targeting within a result row."""
1677 return (
1678 hasattr(other, "name")
1679 and hasattr(self, "name")
1680 and other.name == self.name
1681 )
1683 @HasMemoized.memoized_attribute
1684 def _proxy_key(self) -> Optional[str]:
1685 if self._annotations and "proxy_key" in self._annotations:
1686 return cast(str, self._annotations["proxy_key"])
1688 name = self.key
1689 if not name:
1690 # there's a bit of a seeming contradiction which is that the
1691 # "_non_anon_label" of a column can in fact be an
1692 # "_anonymous_label"; this is when it's on a column that is
1693 # proxying for an anonymous expression in a subquery.
1694 name = self._non_anon_label
1696 if isinstance(name, _anonymous_label):
1697 return None
1698 else:
1699 return name
1701 @HasMemoized.memoized_attribute
1702 def _expression_label(self) -> Optional[str]:
1703 """a suggested label to use in the case that the column has no name,
1704 which should be used if possible as the explicit 'AS <label>'
1705 where this expression would normally have an anon label.
1707 this is essentially mostly what _proxy_key does except it returns
1708 None if the column has a normal name that can be used.
1710 """
1712 if getattr(self, "name", None) is not None:
1713 return None
1714 elif self._annotations and "proxy_key" in self._annotations:
1715 return cast(str, self._annotations["proxy_key"])
1716 else:
1717 return None
1719 def _make_proxy(
1720 self,
1721 selectable: FromClause,
1722 *,
1723 primary_key: ColumnSet,
1724 foreign_keys: Set[KeyedColumnElement[Any]],
1725 name: Optional[str] = None,
1726 key: Optional[str] = None,
1727 name_is_truncatable: bool = False,
1728 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
1729 **kw: Any,
1730 ) -> typing_Tuple[str, ColumnClause[_T]]:
1731 """Create a new :class:`_expression.ColumnElement` representing this
1732 :class:`_expression.ColumnElement` as it appears in the select list of
1733 a descending selectable.
1735 """
1736 if name is None:
1737 name = self._anon_name_label
1738 if key is None:
1739 key = self._proxy_key
1740 else:
1741 key = name
1743 assert key is not None
1745 co: ColumnClause[_T] = ColumnClause(
1746 (
1747 coercions.expect(roles.TruncatedLabelRole, name)
1748 if name_is_truncatable
1749 else name
1750 ),
1751 type_=getattr(self, "type", None),
1752 _selectable=selectable,
1753 )
1755 co._propagate_attrs = selectable._propagate_attrs
1756 if compound_select_cols:
1757 co._proxies = list(compound_select_cols)
1758 else:
1759 co._proxies = [self]
1760 if selectable._is_clone_of is not None:
1761 co._is_clone_of = selectable._is_clone_of.columns.get(key)
1762 return key, co
1764 def cast(self, type_: _TypeEngineArgument[_OPT]) -> Cast[_OPT]:
1765 """Produce a type cast, i.e. ``CAST(<expression> AS <type>)``.
1767 This is a shortcut to the :func:`_expression.cast` function.
1769 .. seealso::
1771 :ref:`tutorial_casts`
1773 :func:`_expression.cast`
1775 :func:`_expression.type_coerce`
1777 """
1778 return Cast(self, type_)
1780 def label(self, name: Optional[str]) -> Label[_T]:
1781 """Produce a column label, i.e. ``<columnname> AS <name>``.
1783 This is a shortcut to the :func:`_expression.label` function.
1785 If 'name' is ``None``, an anonymous label name will be generated.
1787 """
1788 return Label(name, self, self.type)
1790 def _anon_label(
1791 self, seed: Optional[str], add_hash: Optional[int] = None
1792 ) -> _anonymous_label:
1793 while self._is_clone_of is not None:
1794 self = self._is_clone_of
1796 # as of 1.4 anonymous label for ColumnElement uses hash(), not id(),
1797 # as the identifier, because a column and its annotated version are
1798 # the same thing in a SQL statement
1799 hash_value = hash(self)
1801 if add_hash:
1802 # this path is used for disambiguating anon labels that would
1803 # otherwise be the same name for the same element repeated.
1804 # an additional numeric value is factored in for each label.
1806 # shift hash(self) (which is id(self), typically 8 byte integer)
1807 # 16 bits leftward. fill extra add_hash on right
1808 assert add_hash < (2 << 15)
1809 assert seed
1810 hash_value = (hash_value << 16) | add_hash
1812 # extra underscore is added for labels with extra hash
1813 # values, to isolate the "deduped anon" namespace from the
1814 # regular namespace. eliminates chance of these
1815 # manufactured hash values overlapping with regular ones for some
1816 # undefined python interpreter
1817 seed = seed + "_"
1819 if isinstance(seed, _anonymous_label):
1820 # NOTE: the space after the hash is required
1821 return _anonymous_label(f"{seed}%({hash_value} )s")
1823 return _anonymous_label.safe_construct(hash_value, seed or "anon")
1825 @util.memoized_property
1826 def _anon_name_label(self) -> str:
1827 """Provides a constant 'anonymous label' for this ColumnElement.
1829 This is a label() expression which will be named at compile time.
1830 The same label() is returned each time ``anon_label`` is called so
1831 that expressions can reference ``anon_label`` multiple times,
1832 producing the same label name at compile time.
1834 The compiler uses this function automatically at compile time
1835 for expressions that are known to be 'unnamed' like binary
1836 expressions and function calls.
1838 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1839 public and is renamed to _anon_name_label. anon_name exists
1840 for backwards compat
1842 """
1843 name = getattr(self, "name", None)
1844 return self._anon_label(name)
1846 @util.memoized_property
1847 def _anon_key_label(self) -> _anonymous_label:
1848 """Provides a constant 'anonymous key label' for this ColumnElement.
1850 Compare to ``anon_label``, except that the "key" of the column,
1851 if available, is used to generate the label.
1853 This is used when a deduplicating key is placed into the columns
1854 collection of a selectable.
1856 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1857 public and is renamed to _anon_key_label. anon_key_label exists
1858 for backwards compat
1860 """
1861 return self._anon_label(self._proxy_key)
1863 @property
1864 @util.deprecated(
1865 "1.4",
1866 "The :attr:`_expression.ColumnElement.anon_label` attribute is now "
1867 "private, and the public accessor is deprecated.",
1868 )
1869 def anon_label(self) -> str:
1870 return self._anon_name_label
1872 @property
1873 @util.deprecated(
1874 "1.4",
1875 "The :attr:`_expression.ColumnElement.anon_key_label` attribute is "
1876 "now private, and the public accessor is deprecated.",
1877 )
1878 def anon_key_label(self) -> str:
1879 return self._anon_key_label
1881 def _dedupe_anon_label_idx(self, idx: int) -> str:
1882 """label to apply to a column that is anon labeled, but repeated
1883 in the SELECT, so that we have to make an "extra anon" label that
1884 disambiguates it from the previous appearance.
1886 these labels come out like "foo_bar_id__1" and have double underscores
1887 in them.
1889 """
1890 label = getattr(self, "name", None)
1892 # current convention is that if the element doesn't have a
1893 # ".name" (usually because it is not NamedColumn), we try to
1894 # use a "table qualified" form for the "dedupe anon" label,
1895 # based on the notion that a label like
1896 # "CAST(casttest.v1 AS DECIMAL) AS casttest_v1__1" looks better than
1897 # "CAST(casttest.v1 AS DECIMAL) AS anon__1"
1899 if label is None:
1900 return self._dedupe_anon_tq_label_idx(idx)
1901 else:
1902 return self._anon_label(label, add_hash=idx)
1904 @util.memoized_property
1905 def _anon_tq_label(self) -> _anonymous_label:
1906 return self._anon_label(getattr(self, "_tq_label", None))
1908 @util.memoized_property
1909 def _anon_tq_key_label(self) -> _anonymous_label:
1910 return self._anon_label(getattr(self, "_tq_key_label", None))
1912 def _dedupe_anon_tq_label_idx(self, idx: int) -> _anonymous_label:
1913 label = getattr(self, "_tq_label", None) or "anon"
1915 return self._anon_label(label, add_hash=idx)
1918class KeyedColumnElement(ColumnElement[_T]):
1919 """ColumnElement where ``.key`` is non-None."""
1921 _is_keyed_column_element = True
1923 key: str
1926class WrapsColumnExpression(ColumnElement[_T]):
1927 """Mixin that defines a :class:`_expression.ColumnElement`
1928 as a wrapper with special
1929 labeling behavior for an expression that already has a name.
1931 .. versionadded:: 1.4
1933 .. seealso::
1935 :ref:`change_4449`
1938 """
1940 @property
1941 def wrapped_column_expression(self) -> ColumnElement[_T]:
1942 raise NotImplementedError()
1944 @util.non_memoized_property
1945 def _tq_label(self) -> Optional[str]:
1946 wce = self.wrapped_column_expression
1947 if hasattr(wce, "_tq_label"):
1948 return wce._tq_label
1949 else:
1950 return None
1952 @property
1953 def _label(self) -> Optional[str]:
1954 return self._tq_label
1956 @property
1957 def _non_anon_label(self) -> Optional[str]:
1958 return None
1960 @util.non_memoized_property
1961 def _anon_name_label(self) -> str:
1962 wce = self.wrapped_column_expression
1964 # this logic tries to get the WrappedColumnExpression to render
1965 # with "<expr> AS <name>", where "<name>" is the natural name
1966 # within the expression itself. e.g. "CAST(table.foo) AS foo".
1967 if not wce._is_text_clause:
1968 nal = wce._non_anon_label
1969 if nal:
1970 return nal
1971 elif hasattr(wce, "_anon_name_label"):
1972 return wce._anon_name_label
1973 return super()._anon_name_label
1975 def _dedupe_anon_label_idx(self, idx: int) -> str:
1976 wce = self.wrapped_column_expression
1977 nal = wce._non_anon_label
1978 if nal:
1979 return self._anon_label(nal + "_")
1980 else:
1981 return self._dedupe_anon_tq_label_idx(idx)
1983 @property
1984 def _proxy_key(self):
1985 wce = self.wrapped_column_expression
1987 if not wce._is_text_clause:
1988 return wce._proxy_key
1989 return super()._proxy_key
1992class DMLTargetCopy(roles.InElementRole, KeyedColumnElement[_T]):
1993 """Refer to another column's VALUES or SET expression in an INSERT or
1994 UPDATE statement.
1996 See the public-facing :func:`_sql.from_dml_column` constructor for
1997 background.
1999 .. versionadded:: 2.1
2002 """
2004 def __init__(self, column: _OnlyColumnArgument[_T]):
2005 self.column = coercions.expect(roles.ColumnArgumentRole, column)
2006 self.type = self.column.type
2008 __visit_name__ = "dmltargetcopy"
2010 _traverse_internals: _TraverseInternalsType = [
2011 ("column", InternalTraversal.dp_clauseelement),
2012 ]
2015class BindParameter(roles.InElementRole, KeyedColumnElement[_T]):
2016 r"""Represent a "bound expression".
2018 :class:`.BindParameter` is invoked explicitly using the
2019 :func:`.bindparam` function, as in::
2021 from sqlalchemy import bindparam
2023 stmt = select(users_table).where(
2024 users_table.c.name == bindparam("username")
2025 )
2027 Detailed discussion of how :class:`.BindParameter` is used is
2028 at :func:`.bindparam`.
2030 .. seealso::
2032 :func:`.bindparam`
2034 """
2036 __visit_name__ = "bindparam"
2038 _traverse_internals: _TraverseInternalsType = [
2039 ("key", InternalTraversal.dp_anon_name),
2040 ("type", InternalTraversal.dp_type),
2041 ("callable", InternalTraversal.dp_plain_dict),
2042 ("value", InternalTraversal.dp_plain_obj),
2043 ("literal_execute", InternalTraversal.dp_boolean),
2044 ]
2046 key: str
2047 _anon_map_key: Optional[str] = None
2048 type: TypeEngine[_T]
2049 value: Optional[_T]
2051 _is_crud = False
2052 _is_bind_parameter = True
2054 # bindparam implements its own _gen_cache_key() method however
2055 # we check subclasses for this flag, else no cache key is generated
2056 inherit_cache = True
2058 def __init__(
2059 self,
2060 key: Optional[str],
2061 value: Any = _NoArg.NO_ARG,
2062 type_: Optional[_TypeEngineArgument[_T]] = None,
2063 unique: bool = False,
2064 required: Union[bool, Literal[_NoArg.NO_ARG]] = _NoArg.NO_ARG,
2065 quote: Optional[bool] = None,
2066 callable_: Optional[Callable[[], Any]] = None,
2067 expanding: bool = False,
2068 isoutparam: bool = False,
2069 literal_execute: bool = False,
2070 _compared_to_operator: Optional[OperatorType] = None,
2071 _compared_to_type: Optional[TypeEngine[Any]] = None,
2072 _is_crud: bool = False,
2073 ):
2074 if required is _NoArg.NO_ARG:
2075 required = value is _NoArg.NO_ARG and callable_ is None
2076 if value is _NoArg.NO_ARG:
2077 value = None
2079 if quote is not None:
2080 key = quoted_name.construct(key, quote)
2082 if unique:
2083 self.key, self._anon_map_key = (
2084 _anonymous_label.safe_construct_with_key(
2085 id(self),
2086 (
2087 key
2088 if key is not None
2089 and not isinstance(key, _anonymous_label)
2090 else "param"
2091 ),
2092 sanitize_key=True,
2093 )
2094 )
2095 elif key:
2096 self.key = key
2097 else:
2098 self.key, self._anon_map_key = (
2099 _anonymous_label.safe_construct_with_key(id(self), "param")
2100 )
2102 # identifying key that won't change across
2103 # clones, used to identify the bind's logical
2104 # identity
2105 self._identifying_key = self.key
2107 # key that was passed in the first place, used to
2108 # generate new keys
2109 self._orig_key = key or "param"
2111 self.unique = unique
2112 self.value = value
2113 self.callable = callable_
2114 self.isoutparam = isoutparam
2115 self.required = required
2117 # indicate an "expanding" parameter; the compiler sets this
2118 # automatically in the compiler _render_in_expr_w_bindparam method
2119 # for an IN expression
2120 self.expanding = expanding
2122 # this is another hint to help w/ expanding and is typically
2123 # set in the compiler _render_in_expr_w_bindparam method for an
2124 # IN expression
2125 self.expand_op = None
2127 self.literal_execute = literal_execute
2128 if _is_crud:
2129 self._is_crud = True
2131 if type_ is None:
2132 if expanding:
2133 if value:
2134 check_value = value[0]
2135 else:
2136 check_value = type_api._NO_VALUE_IN_LIST
2137 else:
2138 check_value = value
2139 if _compared_to_type is not None:
2140 self.type = _compared_to_type.coerce_compared_value(
2141 _compared_to_operator, check_value
2142 )
2143 else:
2144 self.type = type_api._resolve_value_to_type(check_value)
2145 elif isinstance(type_, type):
2146 self.type = type_()
2147 elif is_tuple_type(type_):
2148 if value:
2149 if expanding:
2150 check_value = value[0]
2151 else:
2152 check_value = value
2153 cast("BindParameter[TupleAny]", self).type = (
2154 type_._resolve_values_to_types(check_value)
2155 )
2156 else:
2157 cast("BindParameter[TupleAny]", self).type = type_
2158 else:
2159 self.type = type_
2161 def _with_value(self, value, maintain_key=False, required=NO_ARG):
2162 """Return a copy of this :class:`.BindParameter` with the given value
2163 set.
2164 """
2165 cloned = self._clone(maintain_key=maintain_key)
2166 cloned.value = value
2167 cloned.callable = None
2168 cloned.required = required if required is not NO_ARG else self.required
2169 if cloned.type is type_api.NULLTYPE:
2170 cloned.type = type_api._resolve_value_to_type(value)
2171 return cloned
2173 @property
2174 def effective_value(self) -> Optional[_T]:
2175 """Return the value of this bound parameter,
2176 taking into account if the ``callable`` parameter
2177 was set.
2179 The ``callable`` value will be evaluated
2180 and returned if present, else ``value``.
2182 """
2183 if self.callable:
2184 # TODO: set up protocol for bind parameter callable
2185 return self.callable() # type: ignore
2186 else:
2187 return self.value
2189 def render_literal_execute(self) -> Self:
2190 """Produce a copy of this bound parameter that will enable the
2191 :paramref:`_sql.BindParameter.literal_execute` flag.
2193 The :paramref:`_sql.BindParameter.literal_execute` flag will
2194 have the effect of the parameter rendered in the compiled SQL
2195 string using ``[POSTCOMPILE]`` form, which is a special form that
2196 is converted to be a rendering of the literal value of the parameter
2197 at SQL execution time. The rationale is to support caching
2198 of SQL statement strings that can embed per-statement literal values,
2199 such as LIMIT and OFFSET parameters, in the final SQL string that
2200 is passed to the DBAPI. Dialects in particular may want to use
2201 this method within custom compilation schemes.
2203 .. versionadded:: 1.4.5
2205 .. seealso::
2207 :ref:`engine_thirdparty_caching`
2209 """
2210 c: Self = ClauseElement._clone(self)
2211 c.literal_execute = True
2212 return c
2214 def _negate_in_binary(self, negated_op, original_op):
2215 if self.expand_op is original_op:
2216 bind = self._clone()
2217 bind.expand_op = negated_op
2218 return bind
2219 else:
2220 return self
2222 def _with_binary_element_type(self, type_: TypeEngine[Any]) -> Self:
2223 c: Self = ClauseElement._clone(self)
2224 c.type = type_
2225 return c
2227 def _clone(self, maintain_key: bool = False, **kw: Any) -> Self:
2228 c: Self = ClauseElement._clone(self, **kw)
2229 # ensure all the BindParameter objects stay in cloned set.
2230 # in #7823, we changed "clone" so that a clone only keeps a reference
2231 # to the "original" element, since for column correspondence, that's
2232 # all we need. However, for BindParam, _cloned_set is used by
2233 # the "cache key bind match" lookup, which means if any of those
2234 # interim BindParameter objects became part of a cache key in the
2235 # cache, we need it. So here, make sure all clones keep carrying
2236 # forward.
2237 c._cloned_set.update(self._cloned_set)
2238 if not maintain_key and self.unique:
2239 c.key, c._anon_map_key = _anonymous_label.safe_construct_with_key(
2240 id(c), c._orig_key or "param", sanitize_key=True
2241 )
2242 return c
2244 def _gen_cache_key(self, anon_map, bindparams):
2245 _gen_cache_ok = self.__class__.__dict__.get("inherit_cache", False)
2247 if not _gen_cache_ok:
2248 if anon_map is not None:
2249 anon_map[NO_CACHE] = True
2250 return None
2252 id_, found = anon_map.get_anon(self)
2253 if found:
2254 return (id_, self.__class__)
2256 if bindparams is not None:
2257 bindparams.append(self)
2259 return (
2260 id_,
2261 self.__class__,
2262 self.type._static_cache_key,
2263 (
2264 anon_map[self._anon_map_key]
2265 if self._anon_map_key is not None
2266 else self.key
2267 ),
2268 self.literal_execute,
2269 )
2271 def _convert_to_unique(self):
2272 if not self.unique:
2273 self.unique = True
2274 self.key, self._anon_map_key = (
2275 _anonymous_label.safe_construct_with_key(
2276 id(self), self._orig_key or "param", sanitize_key=True
2277 )
2278 )
2280 def __getstate__(self):
2281 """execute a deferred value for serialization purposes."""
2283 d = self.__dict__.copy()
2284 v = self.value
2285 if self.callable:
2286 v = self.callable()
2287 d["callable"] = None
2288 d["value"] = v
2289 return d
2291 def __setstate__(self, state):
2292 if state.get("unique", False):
2293 anon_and_key = _anonymous_label.safe_construct_with_key(
2294 id(self), state.get("_orig_key", "param"), sanitize_key=True
2295 )
2296 state["key"], state["_anon_map_key"] = anon_and_key
2297 self.__dict__.update(state)
2299 def __repr__(self):
2300 return "%s(%r, %r, type_=%r)" % (
2301 self.__class__.__name__,
2302 self.key,
2303 self.value,
2304 self.type,
2305 )
2308class TypeClause(DQLDMLClauseElement):
2309 """Handle a type keyword in a SQL statement.
2311 Used by the ``Case`` statement.
2313 """
2315 __visit_name__ = "typeclause"
2317 _traverse_internals: _TraverseInternalsType = [
2318 ("type", InternalTraversal.dp_type)
2319 ]
2320 type: TypeEngine[Any]
2322 def __init__(self, type_: TypeEngine[Any]):
2323 self.type = type_
2326class AbstractTextClause(
2327 roles.DDLConstraintColumnRole,
2328 roles.DDLExpressionRole,
2329 roles.StatementOptionRole,
2330 roles.WhereHavingRole,
2331 roles.OrderByRole,
2332 roles.FromClauseRole,
2333 roles.SelectStatementRole,
2334 roles.InElementRole,
2335 Generative,
2336 ExecutableStatement,
2337 DQLDMLClauseElement,
2338 roles.BinaryElementRole[Any],
2339):
2340 """Base class for textual SQL constructs like TextClause and TString."""
2342 __visit_name__: str
2344 _is_text_clause = True
2345 _is_textual = True
2346 _is_implicitly_boolean = False
2347 _render_label_in_columns_clause = False
2348 _omit_from_statements = False
2349 _is_collection_aggregate = False
2351 @property
2352 def _hide_froms(self) -> Iterable[FromClause]:
2353 return ()
2355 def __and__(self, other):
2356 # support use in select.where(), query.filter()
2357 return and_(self, other)
2359 @property
2360 def _select_iterable(self) -> _SelectIterable:
2361 return (self,)
2363 # help in those cases where text/tstring() is
2364 # interpreted in a column expression situation
2365 key: Optional[str] = None
2366 _label: Optional[str] = None
2368 _allow_label_resolve = False
2370 @property
2371 def type(self) -> TypeEngine[Any]:
2372 return type_api.NULLTYPE
2374 @property
2375 def comparator(self):
2376 return self.type.comparator_factory(self) # type: ignore
2378 def self_group(
2379 self, against: Optional[OperatorType] = None
2380 ) -> Union[Self, Grouping[Any]]:
2381 if against is operators.in_op:
2382 return Grouping(self)
2383 else:
2384 return self
2386 def bindparams(
2387 self,
2388 *binds: BindParameter[Any],
2389 **names_to_values: Any,
2390 ) -> Self:
2391 """Establish the values and/or types of bound parameters within
2392 this :class:`_expression.AbstractTextClause` construct.
2394 This is implemented only for :class:`.TextClause` will raise
2395 ``NotImplementedError`` for :class:`.TString`.
2397 """
2398 raise NotImplementedError()
2400 @util.preload_module("sqlalchemy.sql.selectable")
2401 def columns(
2402 self,
2403 *cols: _OnlyColumnArgument[Any],
2404 **types: _TypeEngineArgument[Any],
2405 ) -> TextualSelect:
2406 r"""Turn this :class:`_expression.AbstractTextClause` object into a
2407 :class:`_expression.TextualSelect`
2408 object that serves the same role as a SELECT
2409 statement.
2411 The :class:`_expression.TextualSelect` is part of the
2412 :class:`_expression.SelectBase`
2413 hierarchy and can be embedded into another statement by using the
2414 :meth:`_expression.TextualSelect.subquery` method to produce a
2415 :class:`.Subquery`
2416 object, which can then be SELECTed from.
2418 This function essentially bridges the gap between an entirely
2419 textual SELECT statement and the SQL expression language concept
2420 of a "selectable"::
2422 from sqlalchemy.sql import column, text
2424 stmt = text("SELECT id, name FROM some_table")
2425 stmt = stmt.columns(column("id"), column("name")).subquery("st")
2427 stmt = (
2428 select(mytable)
2429 .select_from(mytable.join(stmt, mytable.c.name == stmt.c.name))
2430 .where(stmt.c.id > 5)
2431 )
2433 Above, we pass a series of :func:`_expression.column` elements to the
2434 :meth:`_expression.AbstractTextClause.columns` method positionally.
2435 These :func:`_expression.column` elements now become first class
2436 elements upon the :attr:`_expression.TextualSelect.selected_columns`
2437 column collection, which then become part of the :attr:`.Subquery.c`
2438 collection after :meth:`_expression.TextualSelect.subquery` is invoked.
2440 The column expressions we pass to
2441 :meth:`_expression.AbstractTextClause.columns` may also be typed; when
2442 we do so, these :class:`.TypeEngine` objects become the effective
2443 return type of the column, so that SQLAlchemy's result-set-processing
2444 systems may be used on the return values. This is often needed for
2445 types such as date or boolean types, as well as for unicode processing
2446 on some dialect configurations::
2448 stmt = text("SELECT id, name, timestamp FROM some_table")
2449 stmt = stmt.columns(
2450 column("id", Integer),
2451 column("name", Unicode),
2452 column("timestamp", DateTime),
2453 )
2455 for id, name, timestamp in connection.execute(stmt):
2456 print(id, name, timestamp)
2458 As a shortcut to the above syntax, keyword arguments referring to
2459 types alone may be used, if only type conversion is needed::
2461 stmt = text("SELECT id, name, timestamp FROM some_table")
2462 stmt = stmt.columns(id=Integer, name=Unicode, timestamp=DateTime)
2464 for id, name, timestamp in connection.execute(stmt):
2465 print(id, name, timestamp)
2467 The positional form of :meth:`_expression.AbstractTextClause.columns`
2468 also provides the unique feature of **positional column targeting**,
2469 which is particularly useful when using the ORM with complex textual
2470 queries. If we specify the columns from our model to
2471 :meth:`_expression.AbstractTextClause.columns`, the result set will
2472 match to those columns positionally, meaning the name or origin of the
2473 column in the textual SQL doesn't matter::
2475 stmt = text(
2476 "SELECT users.id, addresses.id, users.id, "
2477 "users.name, addresses.email_address AS email "
2478 "FROM users JOIN addresses ON users.id=addresses.user_id "
2479 "WHERE users.id = 1"
2480 ).columns(
2481 User.id,
2482 Address.id,
2483 Address.user_id,
2484 User.name,
2485 Address.email_address,
2486 )
2488 query = (
2489 session.query(User)
2490 .from_statement(stmt)
2491 .options(contains_eager(User.addresses))
2492 )
2494 The :meth:`_expression.AbstractTextClause.columns` method provides a
2495 direct route to calling :meth:`_expression.FromClause.subquery` as well
2496 as :meth:`_expression.SelectBase.cte` against a textual SELECT
2497 statement::
2499 stmt = stmt.columns(id=Integer, name=String).cte("st")
2501 stmt = select(sometable).where(sometable.c.id == stmt.c.id)
2503 :param \*cols: A series of :class:`_expression.ColumnElement` objects,
2504 typically
2505 :class:`_schema.Column` objects from a :class:`_schema.Table`
2506 or ORM level
2507 column-mapped attributes, representing a set of columns that this
2508 textual string will SELECT from.
2510 :param \**types: A mapping of string names to :class:`.TypeEngine`
2511 type objects indicating the datatypes to use for names that are
2512 SELECTed from the textual string. Prefer to use the ``*cols``
2513 argument as it also indicates positional ordering.
2515 """
2516 selectable = util.preloaded.sql_selectable
2518 input_cols: List[NamedColumn[Any]] = [
2519 coercions.expect(roles.LabeledColumnExprRole, col) for col in cols
2520 ]
2522 positional_input_cols = [
2523 (
2524 ColumnClause(col.key, types.pop(col.key))
2525 if col.key in types
2526 else col
2527 )
2528 for col in input_cols
2529 ]
2530 keyed_input_cols: List[NamedColumn[Any]] = [
2531 ColumnClause(key, type_) for key, type_ in types.items()
2532 ]
2534 elem = selectable.TextualSelect.__new__(selectable.TextualSelect)
2535 elem._init(
2536 self,
2537 positional_input_cols + keyed_input_cols,
2538 positional=bool(positional_input_cols) and not keyed_input_cols,
2539 )
2540 return elem
2543class TextClause(AbstractTextClause, inspection.Inspectable["TextClause"]):
2544 """Represent a literal SQL text fragment.
2546 E.g.::
2548 from sqlalchemy import text
2550 t = text("SELECT * FROM users")
2551 result = connection.execute(t)
2553 The :class:`_expression.TextClause` construct is produced using the
2554 :func:`_expression.text`
2555 function; see that function for full documentation.
2557 .. seealso::
2559 :func:`_expression.text`
2561 """
2563 __visit_name__ = "textclause"
2565 _traverse_internals: _TraverseInternalsType = [
2566 ("_bindparams", InternalTraversal.dp_string_clauseelement_dict),
2567 ("text", InternalTraversal.dp_string),
2568 ] + ExecutableStatement._executable_traverse_internals
2570 _bind_params_regex = re.compile(r"(?<![:\w\x5c]):(\w+)(?!:)", re.UNICODE)
2572 @property
2573 def _is_star(self) -> bool: # type: ignore[override]
2574 return self.text == "*"
2576 def __init__(self, text: str):
2577 self._bindparams: Dict[str, BindParameter[Any]] = {}
2579 def repl(m):
2580 self._bindparams[m.group(1)] = BindParameter(m.group(1))
2581 return ":%s" % m.group(1)
2583 # scan the string and search for bind parameter names, add them
2584 # to the list of bindparams
2585 self.text = self._bind_params_regex.sub(repl, text)
2587 @_generative
2588 def bindparams(
2589 self,
2590 *binds: BindParameter[Any],
2591 **names_to_values: Any,
2592 ) -> Self:
2593 """Establish the values and/or types of bound parameters within
2594 this :class:`_expression.TextClause` construct.
2596 Given a text construct such as::
2598 from sqlalchemy import text
2600 stmt = text(
2601 "SELECT id, name FROM user WHERE name=:name AND timestamp=:timestamp"
2602 )
2604 the :meth:`_expression.TextClause.bindparams`
2605 method can be used to establish
2606 the initial value of ``:name`` and ``:timestamp``,
2607 using simple keyword arguments::
2609 stmt = stmt.bindparams(
2610 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2611 )
2613 Where above, new :class:`.BindParameter` objects
2614 will be generated with the names ``name`` and ``timestamp``, and
2615 values of ``jack`` and ``datetime.datetime(2012, 10, 8, 15, 12, 5)``,
2616 respectively. The types will be
2617 inferred from the values given, in this case :class:`.String` and
2618 :class:`.DateTime`.
2620 When specific typing behavior is needed, the positional ``*binds``
2621 argument can be used in which to specify :func:`.bindparam` constructs
2622 directly. These constructs must include at least the ``key``
2623 argument, then an optional value and type::
2625 from sqlalchemy import bindparam
2627 stmt = stmt.bindparams(
2628 bindparam("name", value="jack", type_=String),
2629 bindparam("timestamp", type_=DateTime),
2630 )
2632 Above, we specified the type of :class:`.DateTime` for the
2633 ``timestamp`` bind, and the type of :class:`.String` for the ``name``
2634 bind. In the case of ``name`` we also set the default value of
2635 ``"jack"``.
2637 Additional bound parameters can be supplied at statement execution
2638 time, e.g.::
2640 result = connection.execute(
2641 stmt, timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2642 )
2644 The :meth:`_expression.TextClause.bindparams`
2645 method can be called repeatedly,
2646 where it will reuse existing :class:`.BindParameter` objects to add
2647 new information. For example, we can call
2648 :meth:`_expression.TextClause.bindparams`
2649 first with typing information, and a
2650 second time with value information, and it will be combined::
2652 stmt = text(
2653 "SELECT id, name FROM user WHERE name=:name "
2654 "AND timestamp=:timestamp"
2655 )
2656 stmt = stmt.bindparams(
2657 bindparam("name", type_=String), bindparam("timestamp", type_=DateTime)
2658 )
2659 stmt = stmt.bindparams(
2660 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2661 )
2663 The :meth:`_expression.TextClause.bindparams`
2664 method also supports the concept of
2665 **unique** bound parameters. These are parameters that are
2666 "uniquified" on name at statement compilation time, so that multiple
2667 :func:`_expression.text`
2668 constructs may be combined together without the names
2669 conflicting. To use this feature, specify the
2670 :paramref:`.BindParameter.unique` flag on each :func:`.bindparam`
2671 object::
2673 stmt1 = text("select id from table where name=:name").bindparams(
2674 bindparam("name", value="name1", unique=True)
2675 )
2676 stmt2 = text("select id from table where name=:name").bindparams(
2677 bindparam("name", value="name2", unique=True)
2678 )
2680 union = union_all(stmt1.columns(column("id")), stmt2.columns(column("id")))
2682 The above statement will render as:
2684 .. sourcecode:: sql
2686 select id from table where name=:name_1
2687 UNION ALL select id from table where name=:name_2
2689 """ # noqa: E501
2690 self._bindparams = new_params = self._bindparams.copy()
2692 for bind in binds:
2693 try:
2694 # the regex used for text() currently will not match
2695 # a unique/anonymous key in any case, so use the _orig_key
2696 # so that a text() construct can support unique parameters
2697 existing = new_params[bind._orig_key]
2698 except KeyError as err:
2699 raise exc.ArgumentError(
2700 "This text() construct doesn't define a "
2701 "bound parameter named %r" % bind._orig_key
2702 ) from err
2703 else:
2704 new_params[existing._orig_key] = bind
2706 for key, value in names_to_values.items():
2707 try:
2708 existing = new_params[key]
2709 except KeyError as err:
2710 raise exc.ArgumentError(
2711 "This text() construct doesn't define a "
2712 "bound parameter named %r" % key
2713 ) from err
2714 else:
2715 new_params[key] = existing._with_value(value, required=False)
2716 return self
2718 @property
2719 def type(self) -> TypeEngine[Any]:
2720 return type_api.NULLTYPE
2722 @property
2723 def comparator(self):
2724 # TODO: this seems wrong, it seems like we might not
2725 # be using this method.
2726 return self.type.comparator_factory(self) # type: ignore
2728 def self_group(
2729 self, against: Optional[OperatorType] = None
2730 ) -> Union[Self, Grouping[Any]]:
2731 if against is operators.in_op:
2732 return Grouping(self)
2733 else:
2734 return self
2737class TString(AbstractTextClause, inspection.Inspectable["TString"]):
2738 """Represent a SQL template string using Python 3.14+ t-strings.
2740 E.g.::
2742 from sqlalchemy import tstring, column
2744 a = 5
2745 b = 10
2746 stmt = tstring(t"select {a}, {b}")
2747 result = connection.execute(stmt)
2749 The :class:`_expression.TString` construct is produced using the
2750 :func:`_expression.tstring` function; see that function for full
2751 documentation.
2753 .. versionadded:: 2.1
2755 .. seealso::
2757 :func:`_expression.tstring`
2759 """
2761 __visit_name__ = "tstring"
2763 _traverse_internals: _TraverseInternalsType = [
2764 ("parts", InternalTraversal.dp_clauseelement_list)
2765 ] + ExecutableStatement._executable_traverse_internals
2767 @property
2768 def _is_star(self) -> bool: # type: ignore[override]
2769 return (
2770 len(self.parts) == 1
2771 and isinstance(self.parts[0], TextClause)
2772 and self.parts[0]._is_star
2773 )
2775 def __init__(self, template: Template):
2776 """Construct a :class:`_expression.TString` from a Python 3.14+
2777 template string.
2779 :param template: a Python 3.14+ template string (t-string) that
2780 contains SQL fragments and Python expressions to be interpolated.
2782 """
2783 self.parts: List[ClauseElement] = []
2785 if not isinstance(template, Template):
2786 raise exc.ArgumentError("pep-750 Tstring (e.g. t'...') expected")
2788 for part in template:
2789 if isinstance(part, str):
2790 self.parts.append(TextClause(part))
2791 else:
2792 assert hasattr(part, "value")
2793 self.parts.append(
2794 coercions.expect(roles.TStringElementRole, part.value)
2795 )
2797 def bindparams(
2798 self,
2799 *binds: BindParameter[Any],
2800 **names_to_values: Any,
2801 ) -> Self:
2802 """Not supported for TString constructs.
2804 TString constructs do not support .bindparams(). Bind parameters
2805 are automatically created from interpolated values.
2807 """
2808 raise NotImplementedError(
2809 "TString constructs do not support .bindparams(). "
2810 "Bind parameters are automatically created "
2811 "from interpolated values."
2812 )
2815class Null(SingletonConstant, roles.ConstExprRole[None], ColumnElement[None]):
2816 """Represent the NULL keyword in a SQL statement.
2818 :class:`.Null` is accessed as a constant via the
2819 :func:`.null` function.
2821 """
2823 __visit_name__ = "null"
2825 _traverse_internals: _TraverseInternalsType = []
2826 _singleton: Null
2828 if not TYPE_CHECKING:
2830 @util.memoized_property
2831 def type(self) -> TypeEngine[_T]: # noqa: A001
2832 return type_api.NULLTYPE
2834 @classmethod
2835 def _instance(cls) -> Null:
2836 """Return a constant :class:`.Null` construct."""
2838 return Null._singleton
2841Null._create_singleton()
2844class False_(
2845 SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]
2846):
2847 """Represent the ``false`` keyword, or equivalent, in a SQL statement.
2849 :class:`.False_` is accessed as a constant via the
2850 :func:`.false` function.
2852 """
2854 __visit_name__ = "false"
2855 _traverse_internals: _TraverseInternalsType = []
2856 _singleton: False_
2858 if not TYPE_CHECKING:
2860 @util.memoized_property
2861 def type(self) -> TypeEngine[_T]: # noqa: A001
2862 return type_api.BOOLEANTYPE
2864 def _negate(self) -> True_:
2865 return True_._singleton
2867 @classmethod
2868 def _instance(cls) -> False_:
2869 return False_._singleton
2872False_._create_singleton()
2875class True_(SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]):
2876 """Represent the ``true`` keyword, or equivalent, in a SQL statement.
2878 :class:`.True_` is accessed as a constant via the
2879 :func:`.true` function.
2881 """
2883 __visit_name__ = "true"
2885 _traverse_internals: _TraverseInternalsType = []
2886 _singleton: True_
2888 if not TYPE_CHECKING:
2890 @util.memoized_property
2891 def type(self) -> TypeEngine[_T]: # noqa: A001
2892 return type_api.BOOLEANTYPE
2894 def _negate(self) -> False_:
2895 return False_._singleton
2897 @classmethod
2898 def _ifnone(
2899 cls, other: Optional[ColumnElement[Any]]
2900 ) -> ColumnElement[Any]:
2901 if other is None:
2902 return cls._instance()
2903 else:
2904 return other
2906 @classmethod
2907 def _instance(cls) -> True_:
2908 return True_._singleton
2911True_._create_singleton()
2914class ElementList(DQLDMLClauseElement):
2915 """Describe a list of clauses that will be space separated.
2917 This is a minimal version of :class:`.ClauseList` which is used by
2918 the :class:`.HasSyntaxExtension` class. It does not do any coercions
2919 so should be used internally only.
2921 .. versionadded:: 2.1
2923 """
2925 __visit_name__ = "element_list"
2927 _traverse_internals: _TraverseInternalsType = [
2928 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2929 ]
2931 clauses: typing_Tuple[ClauseElement, ...]
2933 def __init__(self, clauses: Sequence[ClauseElement]):
2934 self.clauses = tuple(clauses)
2937class OrderByList(
2938 roles.OrderByRole,
2939 operators.OrderingOperators,
2940 DQLDMLClauseElement,
2941):
2942 """Describe a list of clauses that will be comma separated to nest
2943 within an ORDER BY.
2945 .. versionadded:: 2.1
2947 """
2949 __visit_name__ = "order_by_list"
2951 _traverse_internals: _TraverseInternalsType = [
2952 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2953 ]
2955 clauses: List[ColumnElement[Any]]
2957 def __init__(
2958 self,
2959 clauses: Iterable[Union[OrderByList, _ColumnExpressionArgument[Any]]],
2960 ):
2961 text_converter_role: Type[roles.SQLRole] = roles.ByOfRole
2962 self._text_converter_role = text_converter_role
2964 self.clauses = [
2965 coercions.expect(
2966 text_converter_role, clause, apply_propagate_attrs=self
2967 )
2968 for clause in clauses
2969 ]
2971 def __iter__(self) -> Iterator[ColumnElement[Any]]:
2972 return iter(self.clauses)
2974 def __len__(self) -> int:
2975 return len(self.clauses)
2977 @property
2978 def _select_iterable(self) -> _SelectIterable:
2979 return itertools.chain.from_iterable(
2980 [elem._select_iterable for elem in self.clauses]
2981 )
2983 @util.ro_non_memoized_property
2984 def _from_objects(self) -> List[FromClause]:
2985 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
2987 def self_group(
2988 self, against: Optional[OperatorType] = None
2989 ) -> Union[Self, Grouping[Any]]:
2990 return self
2992 def desc(self) -> OrderByList:
2993 return OrderByList([e.desc() for e in self.clauses])
2995 def asc(self) -> OrderByList:
2996 return OrderByList([e.asc() for e in self.clauses])
2998 def nulls_first(self) -> OrderByList:
2999 return OrderByList([e.nulls_first() for e in self.clauses])
3001 def nulls_last(self) -> OrderByList:
3002 return OrderByList([e.nulls_last() for e in self.clauses])
3005class ClauseList(
3006 roles.InElementRole,
3007 roles.OrderByRole,
3008 roles.ColumnsClauseRole,
3009 roles.DMLColumnRole,
3010 DQLDMLClauseElement,
3011):
3012 """Describe a list of clauses, separated by an operator.
3014 By default, is comma-separated, such as a column listing.
3016 """
3018 __visit_name__ = "clauselist"
3020 # Used by ORM context.py to identify ClauseList objects in legacy
3021 # composite attribute queries (see test_query_cols_legacy test)
3022 _is_clause_list = True
3024 _traverse_internals: _TraverseInternalsType = [
3025 ("clauses", InternalTraversal.dp_clauseelement_list),
3026 ("operator", InternalTraversal.dp_operator),
3027 ]
3029 clauses: List[ColumnElement[Any]]
3031 def __init__(
3032 self,
3033 *clauses: _ColumnExpressionArgument[Any],
3034 operator: OperatorType = operators.comma_op,
3035 group: bool = True,
3036 group_contents: bool = True,
3037 _literal_as_text_role: Type[roles.SQLRole] = roles.WhereHavingRole,
3038 ):
3039 self.operator = operator
3040 self.group = group
3041 self.group_contents = group_contents
3042 clauses_iterator: Iterable[_ColumnExpressionArgument[Any]] = clauses
3043 text_converter_role: Type[roles.SQLRole] = _literal_as_text_role
3044 self._text_converter_role = text_converter_role
3046 if self.group_contents:
3047 self.clauses = [
3048 coercions.expect(
3049 text_converter_role, clause, apply_propagate_attrs=self
3050 ).self_group(against=self.operator)
3051 for clause in clauses_iterator
3052 ]
3053 else:
3054 self.clauses = [
3055 coercions.expect(
3056 text_converter_role, clause, apply_propagate_attrs=self
3057 )
3058 for clause in clauses_iterator
3059 ]
3060 self._is_implicitly_boolean = operators.is_boolean(self.operator)
3062 @classmethod
3063 def _construct_raw(
3064 cls,
3065 operator: OperatorType,
3066 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
3067 ) -> ClauseList:
3068 self = cls.__new__(cls)
3069 self.clauses = list(clauses) if clauses else []
3070 self.group = True
3071 self.operator = operator
3072 self.group_contents = True
3073 self._is_implicitly_boolean = False
3074 return self
3076 def __iter__(self) -> Iterator[ColumnElement[Any]]:
3077 return iter(self.clauses)
3079 def __len__(self) -> int:
3080 return len(self.clauses)
3082 @property
3083 def _select_iterable(self) -> _SelectIterable:
3084 return itertools.chain.from_iterable(
3085 [elem._select_iterable for elem in self.clauses]
3086 )
3088 def append(self, clause):
3089 if self.group_contents:
3090 self.clauses.append(
3091 coercions.expect(self._text_converter_role, clause).self_group(
3092 against=self.operator
3093 )
3094 )
3095 else:
3096 self.clauses.append(
3097 coercions.expect(self._text_converter_role, clause)
3098 )
3100 @util.ro_non_memoized_property
3101 def _from_objects(self) -> List[FromClause]:
3102 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
3104 def self_group(
3105 self, against: Optional[OperatorType] = None
3106 ) -> Union[Self, Grouping[Any]]:
3107 if self.group and operators.is_precedent(self.operator, against):
3108 return Grouping(self)
3109 else:
3110 return self
3113class OperatorExpression(ColumnElement[_T]):
3114 """base for expressions that contain an operator and operands
3116 .. versionadded:: 2.0
3118 """
3120 operator: OperatorType
3121 type: TypeEngine[_T]
3123 group: bool = True
3125 @property
3126 def is_comparison(self):
3127 return operators.is_comparison(self.operator)
3129 def self_group(
3130 self, against: Optional[OperatorType] = None
3131 ) -> Union[Self, Grouping[_T]]:
3132 if (
3133 self.group
3134 and operators.is_precedent(self.operator, against)
3135 or (
3136 # a negate against a non-boolean operator
3137 # doesn't make too much sense but we should
3138 # group for that
3139 against is operators.inv
3140 and not operators.is_boolean(self.operator)
3141 )
3142 ):
3143 return Grouping(self)
3144 else:
3145 return self
3147 @property
3148 def _flattened_operator_clauses(
3149 self,
3150 ) -> typing_Tuple[ColumnElement[Any], ...]:
3151 raise NotImplementedError()
3153 @classmethod
3154 def _construct_for_op(
3155 cls,
3156 left: ColumnElement[Any],
3157 right: ColumnElement[Any],
3158 op: OperatorType,
3159 *,
3160 type_: TypeEngine[_T],
3161 negate: Optional[OperatorType] = None,
3162 modifiers: Optional[Mapping[str, Any]] = None,
3163 ) -> OperatorExpression[_T]:
3164 if operators.is_associative(op):
3165 assert (
3166 negate is None
3167 ), f"negate not supported for associative operator {op}"
3169 multi = False
3170 if getattr(
3171 left, "operator", None
3172 ) is op and type_._compare_type_affinity(left.type):
3173 multi = True
3174 left_flattened = left._flattened_operator_clauses
3175 else:
3176 left_flattened = (left,)
3178 if getattr(
3179 right, "operator", None
3180 ) is op and type_._compare_type_affinity(right.type):
3181 multi = True
3182 right_flattened = right._flattened_operator_clauses
3183 else:
3184 right_flattened = (right,)
3186 if multi:
3187 return ExpressionClauseList._construct_for_list(
3188 op,
3189 type_,
3190 *(left_flattened + right_flattened),
3191 )
3193 if right._is_collection_aggregate:
3194 negate = None
3196 return BinaryExpression(
3197 left, right, op, type_=type_, negate=negate, modifiers=modifiers
3198 )
3201class ExpressionClauseList(OperatorExpression[_T]):
3202 """Describe a list of clauses, separated by an operator,
3203 in a column expression context.
3205 :class:`.ExpressionClauseList` differs from :class:`.ClauseList` in that
3206 it represents a column-oriented DQL expression only, not an open ended
3207 list of anything comma separated.
3209 .. versionadded:: 2.0
3211 """
3213 __visit_name__ = "expression_clauselist"
3215 _traverse_internals: _TraverseInternalsType = [
3216 ("clauses", InternalTraversal.dp_clauseelement_tuple),
3217 ("operator", InternalTraversal.dp_operator),
3218 ]
3220 clauses: typing_Tuple[ColumnElement[Any], ...]
3222 group: bool
3224 def __init__(
3225 self,
3226 operator: OperatorType,
3227 *clauses: _ColumnExpressionArgument[Any],
3228 type_: Optional[_TypeEngineArgument[_T]] = None,
3229 ):
3230 self.operator = operator
3232 self.clauses = tuple(
3233 coercions.expect(
3234 roles.ExpressionElementRole, clause, apply_propagate_attrs=self
3235 )
3236 for clause in clauses
3237 )
3238 self._is_implicitly_boolean = operators.is_boolean(self.operator)
3239 self.type = type_api.to_instance(type_) # type: ignore
3241 @property
3242 def _flattened_operator_clauses(
3243 self,
3244 ) -> typing_Tuple[ColumnElement[Any], ...]:
3245 return self.clauses
3247 def __iter__(self) -> Iterator[ColumnElement[Any]]:
3248 return iter(self.clauses)
3250 def __len__(self) -> int:
3251 return len(self.clauses)
3253 @property
3254 def _select_iterable(self) -> _SelectIterable:
3255 return (self,)
3257 @util.ro_non_memoized_property
3258 def _from_objects(self) -> List[FromClause]:
3259 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
3261 def _append_inplace(self, clause: ColumnElement[Any]) -> None:
3262 self.clauses += (clause,)
3264 @classmethod
3265 def _construct_for_list(
3266 cls,
3267 operator: OperatorType,
3268 type_: TypeEngine[_T],
3269 *clauses: ColumnElement[Any],
3270 group: bool = True,
3271 ) -> ExpressionClauseList[_T]:
3272 self = cls.__new__(cls)
3273 self.group = group
3274 if group:
3275 self.clauses = tuple(
3276 c.self_group(against=operator) for c in clauses
3277 )
3278 else:
3279 self.clauses = clauses
3280 self.operator = operator
3281 self.type = type_
3282 for c in clauses:
3283 if c._propagate_attrs:
3284 self._propagate_attrs = c._propagate_attrs
3285 break
3286 return self
3288 def _negate(self) -> Any:
3289 grouped = self.self_group(against=operators.inv)
3290 assert isinstance(grouped, ColumnElement)
3291 return UnaryExpression(grouped, operator=operators.inv)
3294class BooleanClauseList(ExpressionClauseList[bool]):
3295 __visit_name__ = "expression_clauselist"
3296 inherit_cache = True
3298 def __init__(self, *arg, **kw):
3299 raise NotImplementedError(
3300 "BooleanClauseList has a private constructor"
3301 )
3303 @classmethod
3304 def _process_clauses_for_boolean(
3305 cls,
3306 operator: OperatorType,
3307 continue_on: Any,
3308 skip_on: Any,
3309 clauses: Iterable[ColumnElement[Any]],
3310 ) -> typing_Tuple[int, List[ColumnElement[Any]]]:
3311 has_continue_on = None
3313 convert_clauses = []
3315 against = operators._asbool
3316 lcc = 0
3318 for clause in clauses:
3319 if clause is continue_on:
3320 # instance of continue_on, like and_(x, y, True, z), store it
3321 # if we didn't find one already, we will use it if there
3322 # are no other expressions here.
3323 has_continue_on = clause
3324 elif clause is skip_on:
3325 # instance of skip_on, e.g. and_(x, y, False, z), cancels
3326 # the rest out
3327 convert_clauses = [clause]
3328 lcc = 1
3329 break
3330 else:
3331 if not lcc:
3332 lcc = 1
3333 else:
3334 against = operator
3335 # technically this would be len(convert_clauses) + 1
3336 # however this only needs to indicate "greater than one"
3337 lcc = 2
3338 convert_clauses.append(clause)
3340 if not convert_clauses and has_continue_on is not None:
3341 convert_clauses = [has_continue_on]
3342 lcc = 1
3344 return lcc, [c.self_group(against=against) for c in convert_clauses]
3346 @classmethod
3347 def _construct(
3348 cls,
3349 operator: OperatorType,
3350 continue_on: Any,
3351 skip_on: Any,
3352 initial_clause: Any = _NoArg.NO_ARG,
3353 *clauses: Any,
3354 **kw: Any,
3355 ) -> ColumnElement[Any]:
3356 if initial_clause is _NoArg.NO_ARG:
3357 # no elements period. deprecated use case. return an empty
3358 # ClauseList construct that generates nothing unless it has
3359 # elements added to it.
3360 name = operator.__name__
3362 util.warn_deprecated(
3363 f"Invoking {name}() without arguments is deprecated, and "
3364 f"will be disallowed in a future release. For an empty "
3365 f"""{name}() construct, use '{name}({
3366 'true()' if continue_on is True_._singleton else 'false()'
3367 }, *args)' """
3368 f"""or '{name}({
3369 'True' if continue_on is True_._singleton else 'False'
3370 }, *args)'.""",
3371 version="1.4",
3372 )
3373 return cls._construct_raw(operator)
3375 lcc, convert_clauses = cls._process_clauses_for_boolean(
3376 operator,
3377 continue_on,
3378 skip_on,
3379 [
3380 coercions.expect(roles.WhereHavingRole, clause)
3381 for clause in util.coerce_generator_arg(
3382 (initial_clause,) + clauses
3383 )
3384 ],
3385 )
3387 if lcc > 1:
3388 # multiple elements. Return regular BooleanClauseList
3389 # which will link elements against the operator.
3391 flattened_clauses = itertools.chain.from_iterable(
3392 (
3393 (c for c in to_flat._flattened_operator_clauses)
3394 if getattr(to_flat, "operator", None) is operator
3395 else (to_flat,)
3396 )
3397 for to_flat in convert_clauses
3398 )
3400 return cls._construct_raw(operator, flattened_clauses) # type: ignore # noqa: E501
3401 else:
3402 assert lcc
3403 # just one element. return it as a single boolean element,
3404 # not a list and discard the operator.
3405 return convert_clauses[0]
3407 @classmethod
3408 def _construct_for_whereclause(
3409 cls, clauses: Iterable[ColumnElement[Any]]
3410 ) -> Optional[ColumnElement[bool]]:
3411 operator, continue_on, skip_on = (
3412 operators.and_,
3413 True_._singleton,
3414 False_._singleton,
3415 )
3417 lcc, convert_clauses = cls._process_clauses_for_boolean(
3418 operator,
3419 continue_on,
3420 skip_on,
3421 clauses, # these are assumed to be coerced already
3422 )
3424 if lcc > 1:
3425 # multiple elements. Return regular BooleanClauseList
3426 # which will link elements against the operator.
3427 return cls._construct_raw(operator, convert_clauses)
3428 elif lcc == 1:
3429 # just one element. return it as a single boolean element,
3430 # not a list and discard the operator.
3431 return convert_clauses[0]
3432 else:
3433 return None
3435 @classmethod
3436 def _construct_raw(
3437 cls,
3438 operator: OperatorType,
3439 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
3440 ) -> BooleanClauseList:
3441 self = cls.__new__(cls)
3442 self.clauses = tuple(clauses) if clauses else ()
3443 self.group = True
3444 self.operator = operator
3445 self.type = type_api.BOOLEANTYPE
3446 self._is_implicitly_boolean = True
3447 return self
3449 @classmethod
3450 def and_(
3451 cls,
3452 initial_clause: Union[
3453 Literal[True], _ColumnExpressionArgument[bool], _NoArg
3454 ] = _NoArg.NO_ARG,
3455 *clauses: _ColumnExpressionArgument[bool],
3456 ) -> ColumnElement[bool]:
3457 r"""Produce a conjunction of expressions joined by ``AND``.
3459 See :func:`_sql.and_` for full documentation.
3460 """
3461 return cls._construct(
3462 operators.and_,
3463 True_._singleton,
3464 False_._singleton,
3465 initial_clause,
3466 *clauses,
3467 )
3469 @classmethod
3470 def or_(
3471 cls,
3472 initial_clause: Union[
3473 Literal[False], _ColumnExpressionArgument[bool], _NoArg
3474 ] = _NoArg.NO_ARG,
3475 *clauses: _ColumnExpressionArgument[bool],
3476 ) -> ColumnElement[bool]:
3477 """Produce a conjunction of expressions joined by ``OR``.
3479 See :func:`_sql.or_` for full documentation.
3480 """
3481 return cls._construct(
3482 operators.or_,
3483 False_._singleton,
3484 True_._singleton,
3485 initial_clause,
3486 *clauses,
3487 )
3489 @property
3490 def _select_iterable(self) -> _SelectIterable:
3491 return (self,)
3493 def self_group(
3494 self, against: Optional[OperatorType] = None
3495 ) -> Union[Self, Grouping[bool]]:
3496 if not self.clauses:
3497 return self
3498 else:
3499 return super().self_group(against=against)
3502and_ = BooleanClauseList.and_
3503or_ = BooleanClauseList.or_
3506class Tuple(ClauseList, ColumnElement[TupleAny]):
3507 """Represent a SQL tuple."""
3509 __visit_name__ = "tuple"
3511 _traverse_internals: _TraverseInternalsType = (
3512 ClauseList._traverse_internals + []
3513 )
3515 type: TupleType
3517 @util.preload_module("sqlalchemy.sql.sqltypes")
3518 def __init__(
3519 self,
3520 *clauses: _ColumnExpressionArgument[Any],
3521 types: Optional[Sequence[_TypeEngineArgument[Any]]] = None,
3522 ):
3523 sqltypes = util.preloaded.sql_sqltypes
3525 if types is None:
3526 init_clauses: List[ColumnElement[Any]] = [
3527 coercions.expect(roles.ExpressionElementRole, c)
3528 for c in clauses
3529 ]
3530 else:
3531 if len(types) != len(clauses):
3532 raise exc.ArgumentError(
3533 "Wrong number of elements for %d-tuple: %r "
3534 % (len(types), clauses)
3535 )
3536 init_clauses = [
3537 coercions.expect(
3538 roles.ExpressionElementRole,
3539 c,
3540 type_=typ if not typ._isnull else None,
3541 )
3542 for typ, c in zip(types, clauses)
3543 ]
3545 self.type = sqltypes.TupleType(*[arg.type for arg in init_clauses])
3546 super().__init__(*init_clauses)
3548 @property
3549 def _select_iterable(self) -> _SelectIterable:
3550 return (self,)
3552 def _bind_param(self, operator, obj, type_=None, expanding=False):
3553 if expanding:
3554 return BindParameter(
3555 None,
3556 value=obj,
3557 _compared_to_operator=operator,
3558 unique=True,
3559 expanding=True,
3560 type_=type_,
3561 _compared_to_type=self.type,
3562 )
3563 else:
3564 return Tuple(
3565 *[
3566 BindParameter(
3567 None,
3568 o,
3569 _compared_to_operator=operator,
3570 _compared_to_type=compared_to_type,
3571 unique=True,
3572 type_=type_,
3573 )
3574 for o, compared_to_type in zip(obj, self.type.types)
3575 ]
3576 )
3578 def self_group(self, against: Optional[OperatorType] = None) -> Self:
3579 # Tuple is parenthesized by definition.
3580 return self
3583class Case(ColumnElement[_T]):
3584 """Represent a ``CASE`` expression.
3586 :class:`.Case` is produced using the :func:`.case` factory function,
3587 as in::
3589 from sqlalchemy import case
3591 stmt = select(users_table).where(
3592 case(
3593 (users_table.c.name == "wendy", "W"),
3594 (users_table.c.name == "jack", "J"),
3595 else_="E",
3596 )
3597 )
3599 Details on :class:`.Case` usage is at :func:`.case`.
3601 .. seealso::
3603 :func:`.case`
3605 """
3607 __visit_name__ = "case"
3609 _traverse_internals: _TraverseInternalsType = [
3610 ("value", InternalTraversal.dp_clauseelement),
3611 ("whens", InternalTraversal.dp_clauseelement_tuples),
3612 ("else_", InternalTraversal.dp_clauseelement),
3613 ]
3615 # for case(), the type is derived from the whens. so for the moment
3616 # users would have to cast() the case to get a specific type
3618 whens: List[typing_Tuple[ColumnElement[bool], ColumnElement[_T]]]
3619 else_: Optional[ColumnElement[_T]]
3620 value: Optional[ColumnElement[Any]]
3622 def __init__(
3623 self,
3624 *whens: Union[
3625 typing_Tuple[_ColumnExpressionArgument[bool], Any],
3626 Mapping[Any, Any],
3627 ],
3628 value: Optional[Any] = None,
3629 else_: Optional[Any] = None,
3630 ):
3631 new_whens: Iterable[Any] = coercions._expression_collection_was_a_list(
3632 "whens", "case", whens
3633 )
3634 try:
3635 new_whens = util.dictlike_iteritems(new_whens)
3636 except TypeError:
3637 pass
3639 self.whens = [
3640 (
3641 coercions.expect(
3642 roles.ExpressionElementRole,
3643 c,
3644 apply_propagate_attrs=self,
3645 ).self_group(),
3646 coercions.expect(roles.ExpressionElementRole, r),
3647 )
3648 for (c, r) in new_whens
3649 ]
3651 if value is None:
3652 self.value = None
3653 else:
3654 self.value = coercions.expect(roles.ExpressionElementRole, value)
3656 if else_ is not None:
3657 self.else_ = coercions.expect(roles.ExpressionElementRole, else_)
3658 else:
3659 self.else_ = None
3661 type_ = next(
3662 (
3663 then.type
3664 # Iterate `whens` in reverse to match previous behaviour
3665 # where type of final element took priority
3666 for *_, then in reversed(self.whens)
3667 if not then.type._isnull
3668 ),
3669 self.else_.type if self.else_ is not None else type_api.NULLTYPE,
3670 )
3671 self.type = cast(_T, type_)
3673 @util.ro_non_memoized_property
3674 def _from_objects(self) -> List[FromClause]:
3675 return list(
3676 itertools.chain(*[x._from_objects for x in self.get_children()])
3677 )
3680class Cast(WrapsColumnExpression[_T]):
3681 """Represent a ``CAST`` expression.
3683 :class:`.Cast` is produced using the :func:`.cast` factory function,
3684 as in::
3686 from sqlalchemy import cast, Numeric
3688 stmt = select(cast(product_table.c.unit_price, Numeric(10, 4)))
3690 Details on :class:`.Cast` usage is at :func:`.cast`.
3692 .. seealso::
3694 :ref:`tutorial_casts`
3696 :func:`.cast`
3698 :func:`.try_cast`
3700 :func:`.type_coerce` - an alternative to CAST that coerces the type
3701 on the Python side only, which is often sufficient to generate the
3702 correct SQL and data coercion.
3704 """
3706 __visit_name__ = "cast"
3708 _traverse_internals: _TraverseInternalsType = [
3709 ("clause", InternalTraversal.dp_clauseelement),
3710 ("type", InternalTraversal.dp_type),
3711 ]
3713 clause: ColumnElement[Any]
3714 type: TypeEngine[_T]
3715 typeclause: TypeClause
3717 def __init__(
3718 self,
3719 expression: _ColumnExpressionArgument[Any],
3720 type_: _TypeEngineArgument[_T],
3721 ):
3722 self.type = type_api.to_instance(type_)
3723 self.clause = coercions.expect(
3724 roles.ExpressionElementRole,
3725 expression,
3726 type_=self.type,
3727 apply_propagate_attrs=self,
3728 )
3729 self.typeclause = TypeClause(self.type)
3731 @util.ro_non_memoized_property
3732 def _from_objects(self) -> List[FromClause]:
3733 return self.clause._from_objects
3735 @property
3736 def wrapped_column_expression(self):
3737 return self.clause
3740class TryCast(Cast[_T]):
3741 """Represent a TRY_CAST expression.
3743 Details on :class:`.TryCast` usage is at :func:`.try_cast`.
3745 .. seealso::
3747 :func:`.try_cast`
3749 :ref:`tutorial_casts`
3750 """
3752 __visit_name__ = "try_cast"
3753 inherit_cache = True
3756class TypeCoerce(WrapsColumnExpression[_T]):
3757 """Represent a Python-side type-coercion wrapper.
3759 :class:`.TypeCoerce` supplies the :func:`_expression.type_coerce`
3760 function; see that function for usage details.
3762 .. seealso::
3764 :func:`_expression.type_coerce`
3766 :func:`.cast`
3768 """
3770 __visit_name__ = "type_coerce"
3772 _traverse_internals: _TraverseInternalsType = [
3773 ("clause", InternalTraversal.dp_clauseelement),
3774 ("type", InternalTraversal.dp_type),
3775 ]
3777 clause: ColumnElement[Any]
3778 type: TypeEngine[_T]
3780 def __init__(
3781 self,
3782 expression: _ColumnExpressionArgument[Any],
3783 type_: _TypeEngineArgument[_T],
3784 ):
3785 self.type = type_api.to_instance(type_)
3786 self.clause = coercions.expect(
3787 roles.ExpressionElementRole,
3788 expression,
3789 type_=self.type,
3790 apply_propagate_attrs=self,
3791 )
3793 @util.ro_non_memoized_property
3794 def _from_objects(self) -> List[FromClause]:
3795 return self.clause._from_objects
3797 @HasMemoized.memoized_attribute
3798 def typed_expression(self):
3799 if isinstance(self.clause, BindParameter):
3800 bp = self.clause._clone()
3801 bp.type = self.type
3802 return bp
3803 else:
3804 return self.clause
3806 @property
3807 def wrapped_column_expression(self):
3808 return self.clause
3810 def self_group(
3811 self, against: Optional[OperatorType] = None
3812 ) -> TypeCoerce[_T]:
3813 grouped = self.clause.self_group(against=against)
3814 if grouped is not self.clause:
3815 return TypeCoerce(grouped, self.type)
3816 else:
3817 return self
3820class Extract(ColumnElement[int]):
3821 """Represent a SQL EXTRACT clause, ``extract(field FROM expr)``."""
3823 __visit_name__ = "extract"
3825 _traverse_internals: _TraverseInternalsType = [
3826 ("expr", InternalTraversal.dp_clauseelement),
3827 ("field", InternalTraversal.dp_string),
3828 ]
3830 expr: ColumnElement[Any]
3831 field: str
3833 def __init__(self, field: str, expr: _ColumnExpressionArgument[Any]):
3834 self.type = type_api.INTEGERTYPE
3835 self.field = field
3836 self.expr = coercions.expect(roles.ExpressionElementRole, expr)
3838 @util.ro_non_memoized_property
3839 def _from_objects(self) -> List[FromClause]:
3840 return self.expr._from_objects
3843class _label_reference(ColumnElement[_T]):
3844 """Wrap a column expression as it appears in a 'reference' context.
3846 This expression is any that includes an _order_by_label_element,
3847 which is a Label, or a DESC / ASC construct wrapping a Label.
3849 The production of _label_reference() should occur when an expression
3850 is added to this context; this includes the ORDER BY or GROUP BY of a
3851 SELECT statement, as well as a few other places, such as the ORDER BY
3852 within an OVER clause.
3854 """
3856 __visit_name__ = "label_reference"
3858 _traverse_internals: _TraverseInternalsType = [
3859 ("element", InternalTraversal.dp_clauseelement)
3860 ]
3862 element: ColumnElement[_T]
3864 def __init__(self, element: ColumnElement[_T]):
3865 self.element = element
3866 self._propagate_attrs = element._propagate_attrs
3868 @util.ro_non_memoized_property
3869 def _from_objects(self) -> List[FromClause]:
3870 return []
3873class _textual_label_reference(ColumnElement[Any]):
3874 __visit_name__ = "textual_label_reference"
3876 _traverse_internals: _TraverseInternalsType = [
3877 ("element", InternalTraversal.dp_string)
3878 ]
3880 def __init__(self, element: str):
3881 self.element = element
3883 @util.memoized_property
3884 def _text_clause(self) -> TextClause:
3885 return TextClause(self.element)
3888class UnaryExpression(ColumnElement[_T]):
3889 """Define a 'unary' expression.
3891 A unary expression has a single column expression
3892 and an operator. The operator can be placed on the left
3893 (where it is called the 'operator') or right (where it is called the
3894 'modifier') of the column expression.
3896 :class:`.UnaryExpression` is the basis for several unary operators
3897 including those used by :func:`.desc`, :func:`.asc`, :func:`.distinct`,
3898 :func:`.nulls_first` and :func:`.nulls_last`.
3900 """
3902 __visit_name__ = "unary"
3904 _traverse_internals: _TraverseInternalsType = [
3905 ("element", InternalTraversal.dp_clauseelement),
3906 ("operator", InternalTraversal.dp_operator),
3907 ("modifier", InternalTraversal.dp_operator),
3908 ]
3910 element: ColumnElement[Any]
3911 operator: Optional[OperatorType]
3912 modifier: Optional[OperatorType]
3914 def __init__(
3915 self,
3916 element: ColumnElement[Any],
3917 *,
3918 operator: Optional[OperatorType] = None,
3919 modifier: Optional[OperatorType] = None,
3920 type_: Optional[_TypeEngineArgument[_T]] = None,
3921 wraps_column_expression: bool = False, # legacy, not used as of 2.0.42
3922 ):
3923 self.operator = operator
3924 self.modifier = modifier
3925 self._propagate_attrs = element._propagate_attrs
3926 self.element = element.self_group(
3927 against=self.operator or self.modifier
3928 )
3930 # if type is None, we get NULLTYPE, which is our _T. But I don't
3931 # know how to get the overloads to express that correctly
3932 self.type = type_api.to_instance(type_) # type: ignore
3934 def _wraps_unnamed_column(self):
3935 ungrouped = self.element._ungroup()
3936 return (
3937 not isinstance(ungrouped, NamedColumn)
3938 or ungrouped._non_anon_label is None
3939 )
3941 @classmethod
3942 def _create_nulls_first(
3943 cls,
3944 column: _ColumnExpressionArgument[_T],
3945 ) -> UnaryExpression[_T]:
3946 return UnaryExpression(
3947 coercions.expect(roles.ByOfRole, column),
3948 modifier=operators.nulls_first_op,
3949 )
3951 @classmethod
3952 def _create_nulls_last(
3953 cls,
3954 column: _ColumnExpressionArgument[_T],
3955 ) -> UnaryExpression[_T]:
3956 return UnaryExpression(
3957 coercions.expect(roles.ByOfRole, column),
3958 modifier=operators.nulls_last_op,
3959 )
3961 @classmethod
3962 def _create_desc(
3963 cls, column: _ColumnExpressionOrStrLabelArgument[_T]
3964 ) -> UnaryExpression[_T]:
3966 return UnaryExpression(
3967 coercions.expect(roles.ByOfRole, column),
3968 modifier=operators.desc_op,
3969 )
3971 @classmethod
3972 def _create_asc(
3973 cls,
3974 column: _ColumnExpressionOrStrLabelArgument[_T],
3975 ) -> UnaryExpression[_T]:
3976 return UnaryExpression(
3977 coercions.expect(roles.ByOfRole, column),
3978 modifier=operators.asc_op,
3979 )
3981 @classmethod
3982 def _create_distinct(
3983 cls,
3984 expr: _ColumnExpressionArgument[_T],
3985 ) -> UnaryExpression[_T]:
3986 col_expr: ColumnElement[_T] = coercions.expect(
3987 roles.ExpressionElementRole, expr
3988 )
3989 return UnaryExpression(
3990 col_expr,
3991 operator=operators.distinct_op,
3992 type_=col_expr.type,
3993 )
3995 @classmethod
3996 def _create_bitwise_not(
3997 cls,
3998 expr: _ColumnExpressionArgument[_T],
3999 ) -> UnaryExpression[_T]:
4000 col_expr: ColumnElement[_T] = coercions.expect(
4001 roles.ExpressionElementRole, expr
4002 )
4003 return UnaryExpression(
4004 col_expr,
4005 operator=operators.bitwise_not_op,
4006 type_=col_expr.type,
4007 )
4009 @property
4010 def _order_by_label_element(self) -> Optional[Label[Any]]:
4011 if operators.is_order_by_modifier(self.modifier):
4012 return self.element._order_by_label_element
4013 else:
4014 return None
4016 @util.ro_non_memoized_property
4017 def _from_objects(self) -> List[FromClause]:
4018 return self.element._from_objects
4020 def _negate(self) -> ColumnElement[Any]:
4021 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
4022 return UnaryExpression(
4023 self.self_group(against=operators.inv),
4024 operator=operators.inv,
4025 type_=type_api.BOOLEANTYPE,
4026 )
4027 else:
4028 return ColumnElement._negate(self)
4030 def self_group(
4031 self, against: Optional[OperatorType] = None
4032 ) -> Union[Self, Grouping[_T]]:
4033 if self.operator and operators.is_precedent(self.operator, against):
4034 return Grouping(self)
4035 else:
4036 return self
4039class CollectionAggregate(UnaryExpression[_T]):
4040 """Forms the basis for right-hand collection operator modifiers
4041 ANY and ALL.
4043 The ANY and ALL keywords are available in different ways on different
4044 backends. On PostgreSQL, they only work for an ARRAY type. On
4045 MySQL, they only work for subqueries.
4047 """
4049 inherit_cache = True
4050 _is_collection_aggregate = True
4052 @classmethod
4053 def _create_any(
4054 cls, expr: _ColumnExpressionArgument[_T]
4055 ) -> CollectionAggregate[bool]:
4056 """create CollectionAggregate for the legacy
4057 ARRAY.Comparator.any() method"""
4058 col_expr: ColumnElement[_T] = coercions.expect(
4059 roles.ExpressionElementRole,
4060 expr,
4061 )
4062 col_expr = col_expr.self_group()
4063 return CollectionAggregate(
4064 col_expr,
4065 operator=operators.any_op,
4066 type_=type_api.BOOLEANTYPE,
4067 )
4069 @classmethod
4070 def _create_all(
4071 cls, expr: _ColumnExpressionArgument[_T]
4072 ) -> CollectionAggregate[bool]:
4073 """create CollectionAggregate for the legacy
4074 ARRAY.Comparator.all() method"""
4075 col_expr: ColumnElement[_T] = coercions.expect(
4076 roles.ExpressionElementRole,
4077 expr,
4078 )
4079 col_expr = col_expr.self_group()
4080 return CollectionAggregate(
4081 col_expr,
4082 operator=operators.all_op,
4083 type_=type_api.BOOLEANTYPE,
4084 )
4086 @util.preload_module("sqlalchemy.sql.sqltypes")
4087 def _bind_param(
4088 self,
4089 operator: operators.OperatorType,
4090 obj: Any,
4091 type_: Optional[TypeEngine[_T]] = None,
4092 expanding: bool = False,
4093 ) -> BindParameter[_T]:
4094 """For new style any_(), all_(), ensure compared literal value
4095 receives appropriate bound parameter type."""
4097 # a CollectionAggregate is specific to ARRAY or int
4098 # only. So for ARRAY case, make sure we use correct element type
4099 sqltypes = util.preloaded.sql_sqltypes
4100 if self.element.type._type_affinity is sqltypes.ARRAY:
4101 compared_to_type = cast(
4102 sqltypes.ARRAY[Any], self.element.type
4103 ).item_type
4104 else:
4105 compared_to_type = self.element.type
4107 return BindParameter(
4108 None,
4109 obj,
4110 _compared_to_operator=operator,
4111 type_=type_,
4112 _compared_to_type=compared_to_type,
4113 unique=True,
4114 expanding=expanding,
4115 )
4117 # operate and reverse_operate are hardwired to
4118 # dispatch onto the type comparator directly, so that we can
4119 # ensure "reversed" behavior.
4120 def operate(
4121 self, op: OperatorType, *other: Any, **kwargs: Any
4122 ) -> ColumnElement[_T]:
4123 if not operators.is_comparison(op):
4124 raise exc.ArgumentError(
4125 "Only comparison operators may be used with ANY/ALL"
4126 )
4127 kwargs["reverse"] = True
4128 return self.comparator.operate(operators.mirror(op), *other, **kwargs)
4130 def reverse_operate(
4131 self, op: OperatorType, other: Any, **kwargs: Any
4132 ) -> ColumnElement[_T]:
4133 # comparison operators should never call reverse_operate
4134 assert not operators.is_comparison(op)
4135 raise exc.ArgumentError(
4136 "Only comparison operators may be used with ANY/ALL"
4137 )
4140class AsBoolean(WrapsColumnExpression[bool], UnaryExpression[bool]):
4141 inherit_cache = True
4143 def __init__(self, element, operator, negate):
4144 self.element = element
4145 self.type = type_api.BOOLEANTYPE
4146 self.operator = operator
4147 self.negate = negate
4148 self.modifier = None
4149 self._is_implicitly_boolean = element._is_implicitly_boolean
4151 @property
4152 def wrapped_column_expression(self):
4153 return self.element
4155 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4156 return self
4158 def _negate(self):
4159 if isinstance(self.element, (True_, False_)):
4160 return self.element._negate()
4161 else:
4162 return AsBoolean(self.element, self.negate, self.operator)
4165class BinaryExpression(OperatorExpression[_T]):
4166 """Represent an expression that is ``LEFT <operator> RIGHT``.
4168 A :class:`.BinaryExpression` is generated automatically
4169 whenever two column expressions are used in a Python binary expression:
4171 .. sourcecode:: pycon+sql
4173 >>> from sqlalchemy.sql import column
4174 >>> column("a") + column("b")
4175 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
4176 >>> print(column("a") + column("b"))
4177 {printsql}a + b
4179 """
4181 __visit_name__ = "binary"
4183 _traverse_internals: _TraverseInternalsType = [
4184 ("left", InternalTraversal.dp_clauseelement),
4185 ("right", InternalTraversal.dp_clauseelement),
4186 ("operator", InternalTraversal.dp_operator),
4187 ("negate", InternalTraversal.dp_operator),
4188 ("modifiers", InternalTraversal.dp_plain_dict),
4189 (
4190 "type",
4191 InternalTraversal.dp_type,
4192 ),
4193 ]
4195 _cache_key_traversal = [
4196 ("left", InternalTraversal.dp_clauseelement),
4197 ("right", InternalTraversal.dp_clauseelement),
4198 ("operator", InternalTraversal.dp_operator),
4199 ("modifiers", InternalTraversal.dp_plain_dict),
4200 # "type" affects JSON CAST operators, so while redundant in most cases,
4201 # is needed for that one
4202 (
4203 "type",
4204 InternalTraversal.dp_type,
4205 ),
4206 ]
4208 _is_implicitly_boolean = True
4209 """Indicates that any database will know this is a boolean expression
4210 even if the database does not have an explicit boolean datatype.
4212 """
4214 left: ColumnElement[Any]
4215 right: ColumnElement[Any]
4216 modifiers: Mapping[str, Any]
4218 def __init__(
4219 self,
4220 left: ColumnElement[Any],
4221 right: ColumnElement[Any],
4222 operator: OperatorType,
4223 type_: Optional[_TypeEngineArgument[_T]] = None,
4224 negate: Optional[OperatorType] = None,
4225 modifiers: Optional[Mapping[str, Any]] = None,
4226 ):
4227 # allow compatibility with libraries that
4228 # refer to BinaryExpression directly and pass strings
4229 if isinstance(operator, str):
4230 operator = operators.custom_op(operator)
4231 self._orig = (left.__hash__(), right.__hash__())
4232 self._propagate_attrs = left._propagate_attrs or right._propagate_attrs
4233 self.left = left.self_group(against=operator)
4234 self.right = right.self_group(against=operator)
4235 self.operator = operator
4237 # if type is None, we get NULLTYPE, which is our _T. But I don't
4238 # know how to get the overloads to express that correctly
4239 self.type = type_api.to_instance(type_) # type: ignore
4241 self.negate = negate
4242 self._is_implicitly_boolean = operators.is_boolean(operator)
4244 if modifiers is None:
4245 self.modifiers = {}
4246 else:
4247 self.modifiers = modifiers
4249 @property
4250 def _flattened_operator_clauses(
4251 self,
4252 ) -> typing_Tuple[ColumnElement[Any], ...]:
4253 return (self.left, self.right)
4255 def __bool__(self):
4256 """Implement Python-side "bool" for BinaryExpression as a
4257 simple "identity" check for the left and right attributes,
4258 if the operator is "eq" or "ne". Otherwise the expression
4259 continues to not support "bool" like all other column expressions.
4261 The rationale here is so that ColumnElement objects can be hashable.
4262 What? Well, suppose you do this::
4264 c1, c2 = column("x"), column("y")
4265 s1 = set([c1, c2])
4267 We do that **a lot**, columns inside of sets is an extremely basic
4268 thing all over the ORM for example.
4270 So what happens if we do this? ::
4272 c1 in s1
4274 Hashing means it will normally use ``__hash__()`` of the object,
4275 but in case of hash collision, it's going to also do ``c1 == c1``
4276 and/or ``c1 == c2`` inside. Those operations need to return a
4277 True/False value. But because we override ``==`` and ``!=``, they're
4278 going to get a BinaryExpression. Hence we implement ``__bool__`` here
4279 so that these comparisons behave in this particular context mostly
4280 like regular object comparisons. Thankfully Python is OK with
4281 that! Otherwise we'd have to use special set classes for columns
4282 (which we used to do, decades ago).
4284 """
4285 if self.operator in (operators.eq, operators.ne):
4286 # this is using the eq/ne operator given int hash values,
4287 # rather than Operator, so that "bool" can be based on
4288 # identity
4289 return self.operator(*self._orig) # type: ignore
4290 else:
4291 raise TypeError("Boolean value of this clause is not defined")
4293 if typing.TYPE_CHECKING:
4295 def __invert__(
4296 self: BinaryExpression[_T],
4297 ) -> BinaryExpression[_T]: ...
4299 @util.ro_non_memoized_property
4300 def _from_objects(self) -> List[FromClause]:
4301 return self.left._from_objects + self.right._from_objects
4303 def _negate(self):
4304 if self.negate is not None:
4305 return BinaryExpression(
4306 self.left,
4307 self.right._negate_in_binary(self.negate, self.operator),
4308 self.negate,
4309 negate=self.operator,
4310 type_=self.type,
4311 modifiers=self.modifiers,
4312 )
4313 else:
4314 return self.self_group()._negate()
4317class Slice(ColumnElement[Any]):
4318 """Represent SQL for a Python array-slice object.
4320 This is not a specific SQL construct at this level, but
4321 may be interpreted by specific dialects, e.g. PostgreSQL.
4323 """
4325 __visit_name__ = "slice"
4327 _traverse_internals: _TraverseInternalsType = [
4328 ("start", InternalTraversal.dp_clauseelement),
4329 ("stop", InternalTraversal.dp_clauseelement),
4330 ("step", InternalTraversal.dp_clauseelement),
4331 ]
4333 def __init__(self, start, stop, step, _name=None):
4334 self.start = coercions.expect(
4335 roles.ExpressionElementRole,
4336 start,
4337 name=_name,
4338 type_=type_api.INTEGERTYPE,
4339 )
4340 self.stop = coercions.expect(
4341 roles.ExpressionElementRole,
4342 stop,
4343 name=_name,
4344 type_=type_api.INTEGERTYPE,
4345 )
4346 self.step = coercions.expect(
4347 roles.ExpressionElementRole,
4348 step,
4349 name=_name,
4350 type_=type_api.INTEGERTYPE,
4351 )
4352 self.type = type_api.NULLTYPE
4354 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4355 assert against is operator.getitem
4356 return self
4359class IndexExpression(BinaryExpression[Any]):
4360 """Represent the class of expressions that are like an "index"
4361 operation."""
4363 inherit_cache = True
4366class GroupedElement(DQLDMLClauseElement):
4367 """Represent any parenthesized expression"""
4369 __visit_name__ = "grouping"
4371 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4372 return self
4374 def _ungroup(self) -> ClauseElement:
4375 raise NotImplementedError()
4378class Grouping(GroupedElement, ColumnElement[_T]):
4379 """Represent a grouping within a column expression"""
4381 _traverse_internals: _TraverseInternalsType = [
4382 ("element", InternalTraversal.dp_clauseelement),
4383 ("type", InternalTraversal.dp_type),
4384 ]
4386 _cache_key_traversal = [
4387 ("element", InternalTraversal.dp_clauseelement),
4388 ]
4390 element: Union[
4391 AbstractTextClause,
4392 ClauseList,
4393 ColumnElement[_T],
4394 CompilerColumnElement,
4395 ]
4397 def __init__(
4398 self,
4399 element: Union[
4400 AbstractTextClause,
4401 ClauseList,
4402 ColumnElement[_T],
4403 CompilerColumnElement,
4404 ],
4405 ):
4406 self.element = element
4408 # nulltype assignment issue
4409 self.type = getattr(element, "type", type_api.NULLTYPE) # type: ignore
4410 self._propagate_attrs = element._propagate_attrs
4412 def _with_binary_element_type(self, type_):
4413 return self.__class__(self.element._with_binary_element_type(type_))
4415 def _ungroup(self) -> ColumnElement[_T]:
4416 assert isinstance(self.element, ColumnElement)
4417 return self.element._ungroup()
4419 @util.memoized_property
4420 def _is_implicitly_boolean(self):
4421 return self.element._is_implicitly_boolean
4423 @util.non_memoized_property
4424 def _tq_label(self) -> Optional[str]:
4425 return (
4426 getattr(self.element, "_tq_label", None) or self._anon_name_label
4427 )
4429 @util.non_memoized_property
4430 def _proxies(self) -> List[ColumnElement[Any]]:
4431 if isinstance(self.element, ColumnElement):
4432 return [self.element]
4433 else:
4434 return []
4436 @util.ro_non_memoized_property
4437 def _from_objects(self) -> List[FromClause]:
4438 return self.element._from_objects
4440 def __getattr__(self, attr):
4441 return getattr(self.element, attr)
4443 def __getstate__(self):
4444 return {"element": self.element, "type": self.type}
4446 def __setstate__(self, state):
4447 self.element = state["element"]
4448 self.type = state["type"]
4450 if TYPE_CHECKING:
4452 def self_group(
4453 self, against: Optional[OperatorType] = None
4454 ) -> Self: ...
4457class _OverrideBinds(Grouping[_T]):
4458 """used by cache_key->_apply_params_to_element to allow compilation /
4459 execution of a SQL element that's been cached, using an alternate set of
4460 bound parameter values.
4462 This is used by the ORM to swap new parameter values into expressions
4463 that are embedded into loader options like with_expression(),
4464 selectinload(). Previously, this task was accomplished using the
4465 .params() method which would perform a deep-copy instead. This deep
4466 copy proved to be too expensive for more complex expressions.
4468 See #11085
4470 """
4472 __visit_name__ = "override_binds"
4474 def __init__(
4475 self,
4476 element: ColumnElement[_T],
4477 bindparams: Sequence[BindParameter[Any]],
4478 replaces_params: Sequence[BindParameter[Any]],
4479 ):
4480 self.element = element
4481 self.translate = {
4482 k.key: v.value for k, v in zip(replaces_params, bindparams)
4483 }
4485 def _gen_cache_key(
4486 self, anon_map: anon_map, bindparams: List[BindParameter[Any]]
4487 ) -> Optional[typing_Tuple[Any, ...]]:
4488 """generate a cache key for the given element, substituting its bind
4489 values for the translation values present."""
4491 existing_bps: List[BindParameter[Any]] = []
4492 ck = self.element._gen_cache_key(anon_map, existing_bps)
4494 bindparams.extend(
4495 (
4496 bp._with_value(
4497 self.translate[bp.key], maintain_key=True, required=False
4498 )
4499 if bp.key in self.translate
4500 else bp
4501 )
4502 for bp in existing_bps
4503 )
4505 return ck
4508_FrameIntTuple = tuple[int | None, int | None]
4511class Over(ColumnElement[_T]):
4512 """Represent an OVER clause.
4514 This is a special operator against a so-called
4515 "window" function, as well as any aggregate function,
4516 which produces results relative to the result set
4517 itself. Most modern SQL backends now support window functions.
4519 """
4521 __visit_name__ = "over"
4523 _traverse_internals: _TraverseInternalsType = [
4524 ("element", InternalTraversal.dp_clauseelement),
4525 ("order_by", InternalTraversal.dp_clauseelement),
4526 ("partition_by", InternalTraversal.dp_clauseelement),
4527 ("range_", InternalTraversal.dp_clauseelement),
4528 ("rows", InternalTraversal.dp_clauseelement),
4529 ("groups", InternalTraversal.dp_clauseelement),
4530 ("exclude", InternalTraversal.dp_string),
4531 ]
4533 order_by: Optional[ClauseList] = None
4534 partition_by: Optional[ClauseList] = None
4536 element: ColumnElement[_T]
4537 """The underlying expression object to which this :class:`.Over`
4538 object refers."""
4540 range_: FrameClause | None
4541 rows: FrameClause | None
4542 groups: FrameClause | None
4543 exclude: str | None
4545 def __init__(
4546 self,
4547 element: ColumnElement[_T],
4548 partition_by: Optional[_ByArgument] = None,
4549 order_by: Optional[_ByArgument] = None,
4550 range_: _FrameIntTuple | FrameClause | None = None,
4551 rows: _FrameIntTuple | FrameClause | None = None,
4552 groups: _FrameIntTuple | FrameClause | None = None,
4553 exclude: str | None = None,
4554 ):
4555 self.element = element
4556 if order_by is not None:
4557 self.order_by = ClauseList(
4558 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4559 )
4560 if partition_by is not None:
4561 self.partition_by = ClauseList(
4562 *util.to_list(partition_by),
4563 _literal_as_text_role=roles.ByOfRole,
4564 )
4566 if sum(item is not None for item in (range_, rows, groups)) > 1:
4567 raise exc.ArgumentError(
4568 "only one of 'rows', 'range_', or 'groups' may be provided"
4569 )
4570 else:
4571 self.range_ = FrameClause._parse(range_, coerce_int=False)
4572 self.rows = FrameClause._parse(rows, coerce_int=True)
4573 self.groups = FrameClause._parse(groups, coerce_int=True)
4575 self.exclude = exclude
4577 if exclude is not None and (
4578 range_ is None and rows is None and groups is None
4579 ):
4580 raise exc.ArgumentError(
4581 "'exclude' requires that one of 'rows', "
4582 "'range_', or 'groups' is also specified"
4583 )
4585 if not TYPE_CHECKING:
4587 @util.memoized_property
4588 def type(self) -> TypeEngine[_T]: # noqa: A001
4589 return self.element.type
4591 @util.ro_non_memoized_property
4592 def _from_objects(self) -> List[FromClause]:
4593 return list(
4594 itertools.chain(
4595 *[
4596 c._from_objects
4597 for c in (self.element, self.partition_by, self.order_by)
4598 if c is not None
4599 ]
4600 )
4601 )
4604class FrameClauseType(Enum):
4605 """Frame clause type enum for FrameClause lower_type and upper_type.
4607 .. versionadded:: 2.1
4609 """
4611 UNBOUNDED = 0
4612 """Produces an "UNBOUNDED PRECEDING" or "UNBOUNDED FOLLOWING" frame
4613 clause depending on the position.
4614 Requires a ``None`` value for the corresponding bound value.
4615 """
4616 CURRENT = 1
4617 """Produces a "CURRENT ROW" frame clause.
4618 Requires a ``None`` value for the corresponding bound value.
4619 """
4620 PRECEDING = 2
4621 """Produces a "PRECEDING" frame clause."""
4622 FOLLOWING = 3
4623 """Produces a "FOLLOWING" frame clause."""
4626_require_none = (
4627 FrameClauseType.CURRENT,
4628 FrameClauseType.UNBOUNDED,
4629)
4632class FrameClause(ClauseElement):
4633 """Indicate the 'rows' 'range' or 'group' field of a window function,
4634 e.g. using :class:`.Over`.
4636 .. versionadded:: 2.1
4638 """
4640 __visit_name__ = "frame_clause"
4642 _traverse_internals: _TraverseInternalsType = [
4643 ("lower_bind", InternalTraversal.dp_clauseelement),
4644 ("upper_bind", InternalTraversal.dp_clauseelement),
4645 ("lower_type", InternalTraversal.dp_plain_obj),
4646 ("upper_type", InternalTraversal.dp_plain_obj),
4647 ]
4649 def __init__(
4650 self,
4651 start: Any,
4652 end: Any,
4653 start_frame_type: FrameClauseType,
4654 end_frame_type: FrameClauseType,
4655 _validate: bool = True,
4656 ) -> None:
4657 """Creates a new FrameClause specifying the bounds of a window frame.
4659 :param start: The start value.
4660 :param end: The end value.
4661 :param start_frame_type: The :class:`FrameClauseType` for the
4662 start value.
4663 :param end_frame_type: The :class:`FrameClauseType` for the end value.
4664 """
4665 self.lower_bind = self._as_literal(start)
4666 self.upper_bind = self._as_literal(end)
4667 self.lower_type = FrameClauseType(start_frame_type)
4668 self.upper_type = FrameClauseType(end_frame_type)
4669 if _validate:
4670 if (
4671 self.lower_type in _require_none
4672 and self.lower_bind is not None
4673 ):
4674 raise exc.ArgumentError(
4675 "Cannot specify a value for start with frame type "
4676 f"{self.lower_type.name}"
4677 )
4678 if (
4679 self.upper_type in _require_none
4680 and self.upper_bind is not None
4681 ):
4682 raise exc.ArgumentError(
4683 "Cannot specify a value for end with frame type "
4684 f"{self.upper_type.name}"
4685 )
4687 @classmethod
4688 def _as_literal(cls, value: Any) -> BindParameter[Any] | None:
4689 if value is None:
4690 return None
4691 elif isinstance(value, int):
4692 return literal(value, type_api.INTEGERTYPE)
4693 elif isinstance(value, BindParameter):
4694 return value
4695 else:
4696 return literal(value) # let the default type resolution occur
4698 @classmethod
4699 def _handle_int(
4700 cls, value: Any | None, coerce_int: bool
4701 ) -> tuple[int | None, FrameClauseType]:
4702 if value is None:
4703 return None, FrameClauseType.UNBOUNDED
4705 if coerce_int:
4706 try:
4707 integer = int(value)
4708 except ValueError as err:
4709 raise exc.ArgumentError(
4710 "Integer or None expected for values in rows/groups frame"
4711 ) from err
4712 elif not isinstance(value, int):
4713 raise exc.ArgumentError(
4714 "When using a tuple to specify a range only integer or none "
4715 "values are allowed in the range frame. To specify a "
4716 "different type use the FrameClause directly."
4717 )
4718 else:
4719 integer = value
4720 if integer == 0:
4721 return None, FrameClauseType.CURRENT
4722 elif integer < 0:
4723 return abs(integer), FrameClauseType.PRECEDING
4724 else:
4725 return integer, FrameClauseType.FOLLOWING
4727 @classmethod
4728 def _parse(
4729 cls,
4730 range_: _FrameIntTuple | FrameClause | None,
4731 coerce_int: bool,
4732 ) -> FrameClause | None:
4733 if range_ is None or isinstance(range_, FrameClause):
4734 return range_
4736 try:
4737 r0, r1 = range_
4738 except (ValueError, TypeError) as ve:
4739 raise exc.ArgumentError(
4740 "2-tuple expected for range/rows/groups"
4741 ) from ve
4743 l_b, l_t = cls._handle_int(r0, coerce_int)
4744 u_b, u_t = cls._handle_int(r1, coerce_int)
4746 return FrameClause(
4747 start=l_b,
4748 end=u_b,
4749 start_frame_type=l_t,
4750 end_frame_type=u_t,
4751 _validate=False,
4752 )
4755class AggregateOrderBy(WrapsColumnExpression[_T]):
4756 """Represent an aggregate ORDER BY expression.
4758 This is a special operator against aggregate functions such as
4759 ``array_agg()``, ``json_arrayagg()`` ``string_agg()``, etc. that provides
4760 for an ORDER BY expression, using a syntax that's compatible with
4761 the backend.
4763 :class:`.AggregateOrderBy` is a generalized version of the
4764 :class:`.WithinGroup` construct, the latter of which always provides a
4765 "WITHIN GROUP (ORDER BY ...)" expression. :class:`.AggregateOrderBy` will
4766 also compile to "WITHIN GROUP (ORDER BY ...)" on backends such as Oracle
4767 and SQL Server that don't have another style of aggregate function
4768 ordering.
4770 .. versionadded:: 2.1
4773 """
4775 __visit_name__ = "aggregateorderby"
4777 _traverse_internals: _TraverseInternalsType = [
4778 ("element", InternalTraversal.dp_clauseelement),
4779 ("order_by", InternalTraversal.dp_clauseelement),
4780 ]
4782 order_by: ClauseList
4784 def __init__(
4785 self,
4786 element: Union[FunctionElement[_T], FunctionFilter[_T]],
4787 *order_by: _ColumnExpressionArgument[Any],
4788 ):
4789 self.element = element
4790 if not order_by:
4791 raise TypeError("at least one ORDER BY element is required")
4792 self.order_by = ClauseList(
4793 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4794 )
4796 if not TYPE_CHECKING:
4798 @util.memoized_property
4799 def type(self) -> TypeEngine[_T]: # noqa: A001
4800 return self.element.type
4802 @property
4803 def wrapped_column_expression(self) -> ColumnElement[_T]:
4804 return self.element
4806 def __reduce__(self):
4807 return self.__class__, (self.element,) + (
4808 tuple(self.order_by) if self.order_by is not None else ()
4809 )
4811 def over(
4812 self,
4813 *,
4814 partition_by: _ByArgument | None = None,
4815 order_by: _ByArgument | None = None,
4816 rows: _FrameIntTuple | FrameClause | None = None,
4817 range_: _FrameIntTuple | FrameClause | None = None,
4818 groups: _FrameIntTuple | FrameClause | None = None,
4819 exclude: str | None = None,
4820 ) -> Over[_T]:
4821 """Produce an OVER clause against this :class:`.WithinGroup`
4822 construct.
4824 This function has the same signature as that of
4825 :meth:`.FunctionElement.over`.
4827 """
4828 return Over(
4829 self,
4830 partition_by=partition_by,
4831 order_by=order_by,
4832 range_=range_,
4833 rows=rows,
4834 groups=groups,
4835 exclude=exclude,
4836 )
4838 @overload
4839 def filter(self) -> Self: ...
4841 @overload
4842 def filter(
4843 self,
4844 __criterion0: _ColumnExpressionArgument[bool],
4845 *criterion: _ColumnExpressionArgument[bool],
4846 ) -> FunctionFilter[_T]: ...
4848 def filter(
4849 self, *criterion: _ColumnExpressionArgument[bool]
4850 ) -> Union[Self, FunctionFilter[_T]]:
4851 """Produce a FILTER clause against this function."""
4852 if not criterion:
4853 return self
4854 return FunctionFilter(self, *criterion)
4856 @util.ro_non_memoized_property
4857 def _from_objects(self) -> List[FromClause]:
4858 return list(
4859 itertools.chain(
4860 *[
4861 c._from_objects
4862 for c in (self.element, self.order_by)
4863 if c is not None
4864 ]
4865 )
4866 )
4869class WithinGroup(AggregateOrderBy[_T]):
4870 """Represent a WITHIN GROUP (ORDER BY) clause.
4872 This is a special operator against so-called
4873 "ordered set aggregate" and "hypothetical
4874 set aggregate" functions, including ``percentile_cont()``,
4875 ``rank()``, ``dense_rank()``, etc.
4877 It's supported only by certain database backends, such as PostgreSQL,
4878 Oracle Database and MS SQL Server.
4880 The :class:`.WithinGroup` construct extracts its type from the
4881 method :meth:`.FunctionElement.within_group_type`. If this returns
4882 ``None``, the function's ``.type`` is used.
4884 """
4886 __visit_name__ = "withingroup"
4887 inherit_cache = True
4889 if not TYPE_CHECKING:
4891 @util.memoized_property
4892 def type(self) -> TypeEngine[_T]: # noqa: A001
4893 wgt = self.element.within_group_type(self)
4894 if wgt is not None:
4895 return wgt
4896 else:
4897 return self.element.type
4900class FunctionFilter(Generative, ColumnElement[_T]):
4901 """Represent a function FILTER clause.
4903 This is a special operator against aggregate and window functions,
4904 which controls which rows are passed to it.
4905 It's supported only by certain database backends.
4907 Invocation of :class:`.FunctionFilter` is via
4908 :meth:`.FunctionElement.filter`::
4910 func.count(1).filter(True)
4912 .. seealso::
4914 :meth:`.FunctionElement.filter`
4916 """
4918 __visit_name__ = "funcfilter"
4920 _traverse_internals: _TraverseInternalsType = [
4921 ("func", InternalTraversal.dp_clauseelement),
4922 ("criterion", InternalTraversal.dp_clauseelement),
4923 ]
4925 criterion: Optional[ColumnElement[bool]] = None
4927 def __init__(
4928 self,
4929 func: Union[FunctionElement[_T], AggregateOrderBy[_T]],
4930 *criterion: _ColumnExpressionArgument[bool],
4931 ):
4932 self.func = func
4933 self.filter.non_generative(self, *criterion) # type: ignore
4935 @_generative
4936 def filter(self, *criterion: _ColumnExpressionArgument[bool]) -> Self:
4937 """Produce an additional FILTER against the function.
4939 This method adds additional criteria to the initial criteria
4940 set up by :meth:`.FunctionElement.filter`.
4942 Multiple criteria are joined together at SQL render time
4943 via ``AND``.
4946 """
4948 for crit in list(criterion):
4949 crit = coercions.expect(roles.WhereHavingRole, crit)
4951 if self.criterion is not None:
4952 self.criterion = self.criterion & crit
4953 else:
4954 self.criterion = crit
4956 return self
4958 def over(
4959 self,
4960 partition_by: _ByArgument | None = None,
4961 order_by: _ByArgument | None = None,
4962 range_: _FrameIntTuple | FrameClause | None = None,
4963 rows: _FrameIntTuple | FrameClause | None = None,
4964 groups: _FrameIntTuple | FrameClause | None = None,
4965 exclude: str | None = None,
4966 ) -> Over[_T]:
4967 """Produce an OVER clause against this filtered function.
4969 Used against aggregate or so-called "window" functions,
4970 for database backends that support window functions.
4972 The expression::
4974 func.rank().filter(MyClass.y > 5).over(order_by="x")
4976 is shorthand for::
4978 from sqlalchemy import over, funcfilter
4980 over(funcfilter(func.rank(), MyClass.y > 5), order_by="x")
4982 See :func:`_expression.over` for a full description.
4984 """
4985 return Over(
4986 self,
4987 partition_by=partition_by,
4988 order_by=order_by,
4989 range_=range_,
4990 rows=rows,
4991 groups=groups,
4992 exclude=exclude,
4993 )
4995 def within_group(
4996 self, *order_by: _ColumnExpressionArgument[Any]
4997 ) -> WithinGroup[_T]:
4998 """Produce a WITHIN GROUP (ORDER BY expr) clause against
4999 this function.
5000 """
5001 return WithinGroup(self, *order_by)
5003 def within_group_type(
5004 self, within_group: WithinGroup[_T]
5005 ) -> Optional[TypeEngine[_T]]:
5006 return None
5008 def self_group(
5009 self, against: Optional[OperatorType] = None
5010 ) -> Union[Self, Grouping[_T]]:
5011 if operators.is_precedent(operators.filter_op, against):
5012 return Grouping(self)
5013 else:
5014 return self
5016 if not TYPE_CHECKING:
5018 @util.memoized_property
5019 def type(self) -> TypeEngine[_T]: # noqa: A001
5020 return self.func.type
5022 @util.ro_non_memoized_property
5023 def _from_objects(self) -> List[FromClause]:
5024 return list(
5025 itertools.chain(
5026 *[
5027 c._from_objects
5028 for c in (self.func, self.criterion)
5029 if c is not None
5030 ]
5031 )
5032 )
5035class NamedColumn(KeyedColumnElement[_T]):
5036 is_literal = False
5037 table: Optional[FromClause] = None
5038 name: str
5039 key: str
5041 def _compare_name_for_result(self, other):
5042 return (hasattr(other, "name") and self.name == other.name) or (
5043 hasattr(other, "_label") and self._label == other._label
5044 )
5046 @util.ro_memoized_property
5047 def description(self) -> str:
5048 return self.name
5050 @HasMemoized.memoized_attribute
5051 def _tq_key_label(self) -> Optional[str]:
5052 """table qualified label based on column key.
5054 for table-bound columns this is <tablename>_<column key/proxy key>;
5056 all other expressions it resolves to key/proxy key.
5058 """
5059 proxy_key = self._proxy_key
5060 if proxy_key and proxy_key != self.name:
5061 return self._gen_tq_label(proxy_key)
5062 else:
5063 return self._tq_label
5065 @HasMemoized.memoized_attribute
5066 def _tq_label(self) -> Optional[str]:
5067 """table qualified label based on column name.
5069 for table-bound columns this is <tablename>_<columnname>; all other
5070 expressions it resolves to .name.
5072 """
5073 return self._gen_tq_label(self.name)
5075 @HasMemoized.memoized_attribute
5076 def _render_label_in_columns_clause(self):
5077 return True
5079 @HasMemoized.memoized_attribute
5080 def _non_anon_label(self):
5081 return self.name
5083 def _gen_tq_label(
5084 self, name: str, dedupe_on_key: bool = True
5085 ) -> Optional[str]:
5086 return name
5088 def _bind_param(
5089 self,
5090 operator: OperatorType,
5091 obj: Any,
5092 type_: Optional[TypeEngine[_T]] = None,
5093 expanding: bool = False,
5094 ) -> BindParameter[_T]:
5095 return BindParameter(
5096 self.key,
5097 obj,
5098 _compared_to_operator=operator,
5099 _compared_to_type=self.type,
5100 type_=type_,
5101 unique=True,
5102 expanding=expanding,
5103 )
5105 def _make_proxy(
5106 self,
5107 selectable: FromClause,
5108 *,
5109 primary_key: ColumnSet,
5110 foreign_keys: Set[KeyedColumnElement[Any]],
5111 name: Optional[str] = None,
5112 key: Optional[str] = None,
5113 name_is_truncatable: bool = False,
5114 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5115 disallow_is_literal: bool = False,
5116 **kw: Any,
5117 ) -> typing_Tuple[str, ColumnClause[_T]]:
5118 c = ColumnClause(
5119 (
5120 coercions.expect(roles.TruncatedLabelRole, name or self.name)
5121 if name_is_truncatable
5122 else (name or self.name)
5123 ),
5124 type_=self.type,
5125 _selectable=selectable,
5126 is_literal=False,
5127 )
5129 c._propagate_attrs = selectable._propagate_attrs
5130 if name is None:
5131 c.key = self.key
5132 if compound_select_cols:
5133 c._proxies = list(compound_select_cols)
5134 else:
5135 c._proxies = [self]
5137 if selectable._is_clone_of is not None:
5138 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5139 return c.key, c
5142_PS = ParamSpec("_PS")
5145class Label(roles.LabeledColumnExprRole[_T], NamedColumn[_T]):
5146 """Represents a column label (AS).
5148 Represent a label, as typically applied to any column-level
5149 element using the ``AS`` sql keyword.
5151 """
5153 __visit_name__ = "label"
5155 _traverse_internals: _TraverseInternalsType = [
5156 ("name", InternalTraversal.dp_anon_name),
5157 ("type", InternalTraversal.dp_type),
5158 ("_element", InternalTraversal.dp_clauseelement),
5159 ]
5161 _cache_key_traversal = [
5162 ("name", InternalTraversal.dp_anon_name),
5163 ("_element", InternalTraversal.dp_clauseelement),
5164 ]
5166 _element: ColumnElement[_T]
5167 name: str
5169 def __init__(
5170 self,
5171 name: Optional[str],
5172 element: _ColumnExpressionArgument[_T],
5173 type_: Optional[_TypeEngineArgument[_T]] = None,
5174 ):
5175 orig_element = element
5176 element = coercions.expect(
5177 roles.ExpressionElementRole,
5178 element,
5179 apply_propagate_attrs=self,
5180 )
5181 while isinstance(element, Label):
5182 # TODO: this is only covered in test_text.py, but nothing
5183 # fails if it's removed. determine rationale
5184 element = element.element
5186 if name:
5187 self.name = name
5188 else:
5189 self.name = _anonymous_label.safe_construct(
5190 id(self), getattr(element, "name", "anon")
5191 )
5192 if isinstance(orig_element, Label):
5193 # TODO: no coverage for this block, again would be in
5194 # test_text.py where the resolve_label concept is important
5195 self._resolve_label = orig_element._label
5197 self.key = self._tq_label = self._tq_key_label = self.name
5198 self._element = element
5200 self.type = (
5201 type_api.to_instance(type_)
5202 if type_ is not None
5203 else self._element.type
5204 )
5206 self._proxies = [element]
5208 def __reduce__(self):
5209 return self.__class__, (self.name, self._element, self.type)
5211 @HasMemoized.memoized_attribute
5212 def _render_label_in_columns_clause(self):
5213 return True
5215 def _bind_param(self, operator, obj, type_=None, expanding=False):
5216 return BindParameter(
5217 None,
5218 obj,
5219 _compared_to_operator=operator,
5220 type_=type_,
5221 _compared_to_type=self.type,
5222 unique=True,
5223 expanding=expanding,
5224 )
5226 @util.memoized_property
5227 def _is_implicitly_boolean(self):
5228 return self.element._is_implicitly_boolean
5230 @HasMemoized.memoized_attribute
5231 def _allow_label_resolve(self):
5232 return self.element._allow_label_resolve
5234 @property
5235 def _order_by_label_element(self):
5236 return self
5238 def as_reference(self) -> _label_reference[_T]:
5239 """refer to this labeled expression in a clause such as GROUP BY,
5240 ORDER BY etc. as the label name itself, without expanding
5241 into the full expression.
5243 .. versionadded:: 2.1
5245 """
5246 return _label_reference(self)
5248 @HasMemoized.memoized_attribute
5249 def element(self) -> ColumnElement[_T]:
5250 return self._element.self_group(against=operators.as_)
5252 def self_group(self, against: Optional[OperatorType] = None) -> Label[_T]:
5253 return self._apply_to_inner(self._element.self_group, against=against)
5255 def _negate(self):
5256 return self._apply_to_inner(self._element._negate)
5258 def _apply_to_inner(
5259 self,
5260 fn: Callable[_PS, ColumnElement[_T]],
5261 *arg: _PS.args,
5262 **kw: _PS.kwargs,
5263 ) -> Label[_T]:
5264 sub_element = fn(*arg, **kw)
5265 if sub_element is not self._element:
5266 return Label(self.name, sub_element, type_=self.type)
5267 else:
5268 return self
5270 @property
5271 def primary_key(self): # type: ignore[override]
5272 return self.element.primary_key
5274 @property
5275 def foreign_keys(self): # type: ignore[override]
5276 return self.element.foreign_keys
5278 def _copy_internals(
5279 self,
5280 *,
5281 clone: _CloneCallableType = _clone,
5282 anonymize_labels: bool = False,
5283 **kw: Any,
5284 ) -> None:
5285 self._reset_memoizations()
5286 self._element = clone(self._element, **kw)
5287 if anonymize_labels:
5288 self.name = _anonymous_label.safe_construct(
5289 id(self), getattr(self.element, "name", "anon")
5290 )
5291 self.key = self._tq_label = self._tq_key_label = self.name
5293 @util.ro_non_memoized_property
5294 def _from_objects(self) -> List[FromClause]:
5295 return self.element._from_objects
5297 def _make_proxy(
5298 self,
5299 selectable: FromClause,
5300 *,
5301 primary_key: ColumnSet,
5302 foreign_keys: Set[KeyedColumnElement[Any]],
5303 name: Optional[str] = None,
5304 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5305 **kw: Any,
5306 ) -> typing_Tuple[str, ColumnClause[_T]]:
5307 name = self.name if not name else name
5309 key, e = self.element._make_proxy(
5310 selectable,
5311 name=name,
5312 disallow_is_literal=True,
5313 name_is_truncatable=isinstance(name, _truncated_label),
5314 compound_select_cols=compound_select_cols,
5315 primary_key=primary_key,
5316 foreign_keys=foreign_keys,
5317 )
5319 # there was a note here to remove this assertion, which was here
5320 # to determine if we later could support a use case where
5321 # the key and name of a label are separate. But I don't know what
5322 # that case was. For now, this is an unexpected case that occurs
5323 # when a label name conflicts with other columns and select()
5324 # is attempting to disambiguate an explicit label, which is not what
5325 # the user would want. See issue #6090.
5326 if key != self.name and not isinstance(self.name, _anonymous_label):
5327 raise exc.InvalidRequestError(
5328 "Label name %s is being renamed to an anonymous label due "
5329 "to disambiguation "
5330 "which is not supported right now. Please use unique names "
5331 "for explicit labels." % (self.name)
5332 )
5334 e._propagate_attrs = selectable._propagate_attrs
5335 e._proxies.append(self)
5336 if self.type is not None:
5337 e.type = self.type
5339 return self.key, e
5342class ColumnClause(
5343 roles.DDLReferredColumnRole,
5344 roles.LabeledColumnExprRole[_T],
5345 roles.StrAsPlainColumnRole,
5346 Immutable,
5347 NamedColumn[_T],
5348):
5349 """Represents a column expression from any textual string.
5351 The :class:`.ColumnClause`, a lightweight analogue to the
5352 :class:`_schema.Column` class, is typically invoked using the
5353 :func:`_expression.column` function, as in::
5355 from sqlalchemy import column
5357 id, name = column("id"), column("name")
5358 stmt = select(id, name).select_from("user")
5360 The above statement would produce SQL like:
5362 .. sourcecode:: sql
5364 SELECT id, name FROM user
5366 :class:`.ColumnClause` is the immediate superclass of the schema-specific
5367 :class:`_schema.Column` object. While the :class:`_schema.Column`
5368 class has all the
5369 same capabilities as :class:`.ColumnClause`, the :class:`.ColumnClause`
5370 class is usable by itself in those cases where behavioral requirements
5371 are limited to simple SQL expression generation. The object has none of
5372 the associations with schema-level metadata or with execution-time
5373 behavior that :class:`_schema.Column` does,
5374 so in that sense is a "lightweight"
5375 version of :class:`_schema.Column`.
5377 Full details on :class:`.ColumnClause` usage is at
5378 :func:`_expression.column`.
5380 .. seealso::
5382 :func:`_expression.column`
5384 :class:`_schema.Column`
5386 """
5388 table: Optional[FromClause]
5389 is_literal: bool
5391 __visit_name__ = "column"
5393 _traverse_internals: _TraverseInternalsType = [
5394 ("name", InternalTraversal.dp_anon_name),
5395 ("type", InternalTraversal.dp_type),
5396 ("table", InternalTraversal.dp_clauseelement),
5397 ("is_literal", InternalTraversal.dp_boolean),
5398 ]
5400 onupdate: Optional[DefaultGenerator] = None
5401 default: Optional[DefaultGenerator] = None
5402 server_default: Optional[FetchedValue] = None
5403 server_onupdate: Optional[FetchedValue] = None
5405 _is_multiparam_column = False
5407 @property
5408 def _is_star(self): # type: ignore[override]
5409 return self.is_literal and self.name == "*"
5411 def __init__(
5412 self,
5413 text: str,
5414 type_: Optional[_TypeEngineArgument[_T]] = None,
5415 is_literal: bool = False,
5416 _selectable: Optional[FromClause] = None,
5417 ):
5418 self.key = self.name = text
5419 self.table = _selectable
5421 # if type is None, we get NULLTYPE, which is our _T. But I don't
5422 # know how to get the overloads to express that correctly
5423 self.type = type_api.to_instance(type_) # type: ignore
5425 self.is_literal = is_literal
5427 def get_children(self, *, column_tables=False, **kw):
5428 # override base get_children() to not return the Table
5429 # or selectable that is parent to this column. Traversals
5430 # expect the columns of tables and subqueries to be leaf nodes.
5431 return []
5433 @property
5434 def entity_namespace(self):
5435 if self.table is not None:
5436 return self.table.entity_namespace
5437 else:
5438 return super().entity_namespace
5440 def _clone(self, detect_subquery_cols=False, **kw):
5441 if (
5442 detect_subquery_cols
5443 and self.table is not None
5444 and self.table._is_subquery
5445 ):
5446 clone = kw.pop("clone")
5447 table = clone(self.table, **kw)
5448 new = table.c.corresponding_column(self)
5449 return new
5451 return super()._clone(**kw)
5453 @HasMemoized_ro_memoized_attribute
5454 def _from_objects(self) -> List[FromClause]:
5455 t = self.table
5456 if t is not None:
5457 return [t]
5458 else:
5459 return []
5461 @HasMemoized.memoized_attribute
5462 def _render_label_in_columns_clause(self):
5463 return self.table is not None
5465 @property
5466 def _ddl_label(self):
5467 return self._gen_tq_label(self.name, dedupe_on_key=False)
5469 def _compare_name_for_result(self, other):
5470 if (
5471 self.is_literal
5472 or self.table is None
5473 or self.table._is_textual
5474 or not hasattr(other, "proxy_set")
5475 or (
5476 isinstance(other, ColumnClause)
5477 and (
5478 other.is_literal
5479 or other.table is None
5480 or other.table._is_textual
5481 )
5482 )
5483 ):
5484 return (hasattr(other, "name") and self.name == other.name) or (
5485 hasattr(other, "_tq_label")
5486 and self._tq_label == other._tq_label
5487 )
5488 else:
5489 return other.proxy_set.intersection(self.proxy_set)
5491 def _gen_tq_label(
5492 self, name: str, dedupe_on_key: bool = True
5493 ) -> Optional[str]:
5494 """generate table-qualified label
5496 for a table-bound column this is <tablename>_<columnname>.
5498 used primarily for LABEL_STYLE_TABLENAME_PLUS_COL
5499 as well as the .columns collection on a Join object.
5501 """
5502 label: str
5503 t = self.table
5504 if self.is_literal:
5505 return None
5506 elif t is not None and is_named_from_clause(t):
5507 if has_schema_attr(t) and t.schema:
5508 label = (
5509 t.schema.replace(".", "_") + "_" + t.name + ("_" + name)
5510 )
5511 else:
5512 assert not TYPE_CHECKING or isinstance(t, NamedFromClause)
5513 label = t.name + ("_" + name)
5515 # propagate name quoting rules for labels.
5516 if is_quoted_name(name) and name.quote is not None:
5517 if is_quoted_name(label):
5518 label.quote = name.quote
5519 else:
5520 label = quoted_name(label, name.quote)
5521 elif is_quoted_name(t.name) and t.name.quote is not None:
5522 # can't get this situation to occur, so let's
5523 # assert false on it for now
5524 assert not isinstance(label, quoted_name)
5525 label = quoted_name(label, t.name.quote)
5527 if dedupe_on_key:
5528 # ensure the label name doesn't conflict with that of an
5529 # existing column. note that this implies that any Column
5530 # must **not** set up its _label before its parent table has
5531 # all of its other Column objects set up. There are several
5532 # tables in the test suite which will fail otherwise; example:
5533 # table "owner" has columns "name" and "owner_name". Therefore
5534 # column owner.name cannot use the label "owner_name", it has
5535 # to be "owner_name_1".
5536 if label in t.c:
5537 _label = label
5538 counter = 1
5539 while _label in t.c:
5540 _label = label + f"_{counter}"
5541 counter += 1
5542 label = _label
5544 return coercions.expect(roles.TruncatedLabelRole, label)
5546 else:
5547 return name
5549 def _make_proxy(
5550 self,
5551 selectable: FromClause,
5552 *,
5553 primary_key: ColumnSet,
5554 foreign_keys: Set[KeyedColumnElement[Any]],
5555 name: Optional[str] = None,
5556 key: Optional[str] = None,
5557 name_is_truncatable: bool = False,
5558 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5559 disallow_is_literal: bool = False,
5560 **kw: Any,
5561 ) -> typing_Tuple[str, ColumnClause[_T]]:
5562 # the "is_literal" flag normally should never be propagated; a proxied
5563 # column is always a SQL identifier and never the actual expression
5564 # being evaluated. however, there is a case where the "is_literal" flag
5565 # might be used to allow the given identifier to have a fixed quoting
5566 # pattern already, so maintain the flag for the proxy unless a
5567 # :class:`.Label` object is creating the proxy. See [ticket:4730].
5568 is_literal = (
5569 not disallow_is_literal
5570 and self.is_literal
5571 and (
5572 # note this does not accommodate for quoted_name differences
5573 # right now
5574 name is None
5575 or name == self.name
5576 )
5577 )
5578 c = self._constructor(
5579 (
5580 coercions.expect(roles.TruncatedLabelRole, name or self.name)
5581 if name_is_truncatable
5582 else (name or self.name)
5583 ),
5584 type_=self.type,
5585 _selectable=selectable,
5586 is_literal=is_literal,
5587 )
5588 c._propagate_attrs = selectable._propagate_attrs
5589 if name is None:
5590 c.key = self.key
5591 if compound_select_cols:
5592 c._proxies = list(compound_select_cols)
5593 else:
5594 c._proxies = [self]
5596 if selectable._is_clone_of is not None:
5597 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5598 return c.key, c
5601class TableValuedColumn(NamedColumn[_T]):
5602 __visit_name__ = "table_valued_column"
5604 _traverse_internals: _TraverseInternalsType = [
5605 ("name", InternalTraversal.dp_anon_name),
5606 ("type", InternalTraversal.dp_type),
5607 ("scalar_alias", InternalTraversal.dp_clauseelement),
5608 ]
5610 def __init__(self, scalar_alias: NamedFromClause, type_: TypeEngine[_T]):
5611 self.scalar_alias = scalar_alias
5612 self.key = self.name = scalar_alias.name
5613 self.type = type_
5615 def _copy_internals(
5616 self, clone: _CloneCallableType = _clone, **kw: Any
5617 ) -> None:
5618 self.scalar_alias = clone(self.scalar_alias, **kw)
5619 self.key = self.name = self.scalar_alias.name
5621 @util.ro_non_memoized_property
5622 def _from_objects(self) -> List[FromClause]:
5623 return [self.scalar_alias]
5626class CollationClause(ColumnElement[str]):
5627 __visit_name__ = "collation"
5629 _traverse_internals: _TraverseInternalsType = [
5630 ("collation", InternalTraversal.dp_string)
5631 ]
5633 @classmethod
5634 @util.preload_module("sqlalchemy.sql.sqltypes")
5635 def _create_collation_expression(
5636 cls, expression: _ColumnExpressionArgument[str], collation: str
5637 ) -> BinaryExpression[str]:
5639 sqltypes = util.preloaded.sql_sqltypes
5641 expr = coercions.expect(roles.ExpressionElementRole[str], expression)
5643 if expr.type._type_affinity is sqltypes.String:
5644 collate_type = expr.type._with_collation(collation)
5645 else:
5646 collate_type = expr.type
5648 return BinaryExpression(
5649 expr,
5650 CollationClause(collation),
5651 operators.collate,
5652 type_=collate_type,
5653 )
5655 def __init__(self, collation):
5656 self.collation = collation
5659class _IdentifiedClause(Executable, ClauseElement):
5660 __visit_name__ = "identified"
5662 def __init__(self, ident):
5663 self.ident = ident
5666class SavepointClause(_IdentifiedClause):
5667 __visit_name__ = "savepoint"
5668 inherit_cache = False
5671class RollbackToSavepointClause(_IdentifiedClause):
5672 __visit_name__ = "rollback_to_savepoint"
5673 inherit_cache = False
5676class ReleaseSavepointClause(_IdentifiedClause):
5677 __visit_name__ = "release_savepoint"
5678 inherit_cache = False
5681class quoted_name(util.MemoizedSlots, str):
5682 """Represent a SQL identifier combined with quoting preferences.
5684 :class:`.quoted_name` is a Python unicode/str subclass which
5685 represents a particular identifier name along with a
5686 ``quote`` flag. This ``quote`` flag, when set to
5687 ``True`` or ``False``, overrides automatic quoting behavior
5688 for this identifier in order to either unconditionally quote
5689 or to not quote the name. If left at its default of ``None``,
5690 quoting behavior is applied to the identifier on a per-backend basis
5691 based on an examination of the token itself.
5693 A :class:`.quoted_name` object with ``quote=True`` is also
5694 prevented from being modified in the case of a so-called
5695 "name normalize" option. Certain database backends, such as
5696 Oracle Database, Firebird, and DB2 "normalize" case-insensitive names
5697 as uppercase. The SQLAlchemy dialects for these backends
5698 convert from SQLAlchemy's lower-case-means-insensitive convention
5699 to the upper-case-means-insensitive conventions of those backends.
5700 The ``quote=True`` flag here will prevent this conversion from occurring
5701 to support an identifier that's quoted as all lower case against
5702 such a backend.
5704 The :class:`.quoted_name` object is normally created automatically
5705 when specifying the name for key schema constructs such as
5706 :class:`_schema.Table`, :class:`_schema.Column`, and others.
5707 The class can also be
5708 passed explicitly as the name to any function that receives a name which
5709 can be quoted. Such as to use the :meth:`_engine.Engine.has_table`
5710 method with
5711 an unconditionally quoted name::
5713 from sqlalchemy import create_engine
5714 from sqlalchemy import inspect
5715 from sqlalchemy.sql import quoted_name
5717 engine = create_engine("oracle+oracledb://some_dsn")
5718 print(inspect(engine).has_table(quoted_name("some_table", True)))
5720 The above logic will run the "has table" logic against the Oracle Database
5721 backend, passing the name exactly as ``"some_table"`` without converting to
5722 upper case.
5724 """
5726 __slots__ = "quote", "lower", "upper"
5728 quote: Optional[bool]
5730 @overload
5731 @classmethod
5732 def construct(cls, value: str, quote: Optional[bool]) -> quoted_name: ...
5734 @overload
5735 @classmethod
5736 def construct(cls, value: None, quote: Optional[bool]) -> None: ...
5738 @classmethod
5739 def construct(
5740 cls, value: Optional[str], quote: Optional[bool]
5741 ) -> Optional[quoted_name]:
5742 if value is None:
5743 return None
5744 else:
5745 return quoted_name(value, quote)
5747 def __new__(cls, value: str, quote: Optional[bool]) -> quoted_name:
5748 assert (
5749 value is not None
5750 ), "use quoted_name.construct() for None passthrough"
5751 if isinstance(value, cls) and (quote is None or value.quote == quote):
5752 return value
5753 self = super().__new__(cls, value)
5755 self.quote = quote
5756 return self
5758 def __reduce__(self):
5759 return quoted_name, (str(self), self.quote)
5761 def _memoized_method_lower(self):
5762 if self.quote:
5763 return self
5764 else:
5765 return str(self).lower()
5767 def _memoized_method_upper(self):
5768 if self.quote:
5769 return self
5770 else:
5771 return str(self).upper()
5774def _find_columns(clause: ClauseElement) -> Set[ColumnClause[Any]]:
5775 """locate Column objects within the given expression."""
5777 cols: Set[ColumnClause[Any]] = set()
5778 traverse(clause, {}, {"column": cols.add})
5779 return cols
5782def _type_from_args(args: Sequence[ColumnElement[_T]]) -> TypeEngine[_T]:
5783 for a in args:
5784 if not a.type._isnull:
5785 return a.type
5786 else:
5787 return type_api.NULLTYPE # type: ignore
5790def _corresponding_column_or_error(fromclause, column, require_embedded=False):
5791 c = fromclause.corresponding_column(
5792 column, require_embedded=require_embedded
5793 )
5794 if c is None:
5795 raise exc.InvalidRequestError(
5796 "Given column '%s', attached to table '%s', "
5797 "failed to locate a corresponding column from table '%s'"
5798 % (column, getattr(column, "table", None), fromclause.description)
5799 )
5800 return c
5803class _memoized_property_but_not_nulltype(
5804 util.memoized_property["TypeEngine[_T]"]
5805):
5806 """memoized property, but dont memoize NullType"""
5808 def __get__(self, obj, cls):
5809 if obj is None:
5810 return self
5811 result = self.fget(obj)
5812 if not result._isnull:
5813 obj.__dict__[self.__name__] = result
5814 return result
5817class AnnotatedColumnElement(Annotated):
5818 _Annotated__element: ColumnElement[Any]
5820 def __init__(self, element, values):
5821 Annotated.__init__(self, element, values)
5822 for attr in (
5823 "comparator",
5824 "_proxy_key",
5825 "_tq_key_label",
5826 "_tq_label",
5827 "_non_anon_label",
5828 "type",
5829 ):
5830 self.__dict__.pop(attr, None)
5831 for attr in ("name", "key", "table"):
5832 if self.__dict__.get(attr, False) is None:
5833 self.__dict__.pop(attr)
5835 def _with_annotations(self, values):
5836 clone = super()._with_annotations(values)
5837 for attr in (
5838 "comparator",
5839 "_proxy_key",
5840 "_tq_key_label",
5841 "_tq_label",
5842 "_non_anon_label",
5843 ):
5844 clone.__dict__.pop(attr, None)
5845 return clone
5847 @util.memoized_property
5848 def name(self):
5849 """pull 'name' from parent, if not present"""
5850 return self._Annotated__element.name
5852 @_memoized_property_but_not_nulltype
5853 def type(self):
5854 """pull 'type' from parent and don't cache if null.
5856 type is routinely changed on existing columns within the
5857 mapped_column() initialization process, and "type" is also consulted
5858 during the creation of SQL expressions. Therefore it can change after
5859 it was already retrieved. At the same time we don't want annotated
5860 objects having overhead when expressions are produced, so continue
5861 to memoize, but only when we have a non-null type.
5863 """
5864 return self._Annotated__element.type
5866 @util.memoized_property
5867 def table(self):
5868 """pull 'table' from parent, if not present"""
5869 return self._Annotated__element.table
5871 @util.memoized_property
5872 def key(self):
5873 """pull 'key' from parent, if not present"""
5874 return self._Annotated__element.key
5876 @util.memoized_property
5877 def info(self) -> _InfoType:
5878 if TYPE_CHECKING:
5879 assert isinstance(self._Annotated__element, Column)
5880 return self._Annotated__element.info
5882 @util.memoized_property
5883 def _anon_name_label(self) -> str:
5884 return self._Annotated__element._anon_name_label
5887class _truncated_label(quoted_name):
5888 """A unicode subclass used to identify symbolic "
5889 "names that may require truncation."""
5891 __slots__ = ()
5893 def __new__(cls, value: str, quote: Optional[bool] = None) -> Any:
5894 quote = getattr(value, "quote", quote)
5895 # return super(_truncated_label, cls).__new__(cls, value, quote, True)
5896 return super().__new__(cls, value, quote)
5898 def __reduce__(self) -> Any:
5899 return self.__class__, (str(self), self.quote)
5901 def apply_map(self, map_: Mapping[str, Any]) -> str:
5902 return self
5905class conv(_truncated_label):
5906 """Mark a string indicating that a name has already been converted
5907 by a naming convention.
5909 This is a string subclass that indicates a name that should not be
5910 subject to any further naming conventions.
5912 E.g. when we create a :class:`.Constraint` using a naming convention
5913 as follows::
5915 m = MetaData(
5916 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5917 )
5918 t = Table(
5919 "t", m, Column("x", Integer), CheckConstraint("x > 5", name="x5")
5920 )
5922 The name of the above constraint will be rendered as ``"ck_t_x5"``.
5923 That is, the existing name ``x5`` is used in the naming convention as the
5924 ``constraint_name`` token.
5926 In some situations, such as in migration scripts, we may be rendering
5927 the above :class:`.CheckConstraint` with a name that's already been
5928 converted. In order to make sure the name isn't double-modified, the
5929 new name is applied using the :func:`_schema.conv` marker. We can
5930 use this explicitly as follows::
5933 m = MetaData(
5934 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5935 )
5936 t = Table(
5937 "t",
5938 m,
5939 Column("x", Integer),
5940 CheckConstraint("x > 5", name=conv("ck_t_x5")),
5941 )
5943 Where above, the :func:`_schema.conv` marker indicates that the constraint
5944 name here is final, and the name will render as ``"ck_t_x5"`` and not
5945 ``"ck_t_ck_t_x5"``
5947 .. seealso::
5949 :ref:`constraint_naming_conventions`
5951 """
5953 __slots__ = ()
5956# for backwards compatibility in case
5957# someone is re-implementing the
5958# _truncated_identifier() sequence in a custom
5959# compiler
5960_generated_label = _truncated_label
5961_anonymous_label_escape = re.compile(r"[%\(\) \$]+")
5964class _anonymous_label(_truncated_label):
5965 """A unicode subclass used to identify anonymously
5966 generated names."""
5968 __slots__ = ()
5970 @classmethod
5971 def safe_construct_with_key(
5972 cls, seed: int | str, body: str, sanitize_key: bool = False
5973 ) -> typing_Tuple[_anonymous_label, str]:
5974 # need to escape chars that interfere with format
5975 # strings in any case, issue #8724
5976 body = _anonymous_label_escape.sub("_", body)
5978 if sanitize_key:
5979 # sanitize_key is then an extra step used by BindParameter
5980 body = body.strip("_")
5982 key = f"{seed} {body.replace('%', '%%')}"
5983 label = _anonymous_label(f"%({key})s")
5984 return label, key
5986 @classmethod
5987 def safe_construct(
5988 cls, seed: int | str, body: str, sanitize_key: bool = False
5989 ) -> _anonymous_label:
5990 # need to escape chars that interfere with format
5991 # strings in any case, issue #8724
5992 body = _anonymous_label_escape.sub("_", body)
5994 if sanitize_key:
5995 # sanitize_key is then an extra step used by BindParameter
5996 body = body.strip("_")
5998 return _anonymous_label(f"%({seed} {body.replace('%', '%%')})s")
6000 def __add__(self, other: str) -> _anonymous_label:
6001 if "%" in other and not isinstance(other, _anonymous_label):
6002 other = str(other).replace("%", "%%")
6003 else:
6004 other = str(other)
6006 return _anonymous_label(
6007 quoted_name(
6008 str.__add__(self, other),
6009 self.quote,
6010 )
6011 )
6013 def __radd__(self, other: str) -> _anonymous_label:
6014 if "%" in other and not isinstance(other, _anonymous_label):
6015 other = str(other).replace("%", "%%")
6016 else:
6017 other = str(other)
6019 return _anonymous_label(
6020 quoted_name(
6021 str.__add__(other, self),
6022 self.quote,
6023 )
6024 )
6026 def apply_map(self, map_: Mapping[str, Any]) -> str:
6027 if self.quote is not None:
6028 # preserve quoting only if necessary
6029 return quoted_name(self % map_, self.quote)
6030 else:
6031 # else skip the constructor call
6032 return self % map_