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 _CovariantTypeEngineArgument
94 from ._typing import _HasDialect
95 from ._typing import _InfoType
96 from ._typing import _OnlyColumnArgument
97 from ._typing import _PropagateAttrsType
98 from ._typing import _TypeEngineArgument
99 from .base import _EntityNamespace
100 from .base import ColumnSet
101 from .cache_key import _CacheKeyTraversalType
102 from .cache_key import CacheKey
103 from .compiler import Compiled
104 from .compiler import SQLCompiler
105 from .functions import FunctionElement
106 from .operators import OperatorType
107 from .schema import Column
108 from .schema import DefaultGenerator
109 from .schema import FetchedValue
110 from .schema import ForeignKey
111 from .selectable import _SelectIterable
112 from .selectable import FromClause
113 from .selectable import NamedFromClause
114 from .selectable import TextualSelect
115 from .sqltypes import TupleType
116 from .type_api import TypeEngine
117 from .visitors import _CloneCallableType
118 from .visitors import _TraverseInternalsType
119 from .visitors import anon_map
120 from ..engine import Connection
121 from ..engine import Dialect
122 from ..engine.interfaces import _CoreMultiExecuteParams
123 from ..engine.interfaces import _CoreSingleExecuteParams
124 from ..engine.interfaces import CacheStats
125 from ..engine.interfaces import CompiledCacheType
126 from ..engine.interfaces import CoreExecuteOptionsParameter
127 from ..engine.interfaces import SchemaTranslateMapType
128 from ..engine.result import Result
131_NUMERIC = Union[float, Decimal]
132_NUMBER = Union[float, int, Decimal]
134_T = TypeVar("_T", bound="Any")
135_T_co = TypeVar("_T_co", bound=Any, covariant=True)
136_OPT = TypeVar("_OPT", bound="Any")
137_NT = TypeVar("_NT", bound="_NUMERIC")
139_NMT = TypeVar("_NMT", bound="_NUMBER")
142@overload
143def literal(
144 value: Any,
145 type_: _CovariantTypeEngineArgument[_T],
146 literal_execute: bool = False,
147) -> BindParameter[_T]: ...
150@overload
151def literal(
152 value: _T,
153 type_: None = None,
154 literal_execute: bool = False,
155) -> BindParameter[_T]: ...
158@overload
159def literal(
160 value: Any,
161 type_: Optional[_TypeEngineArgument[Any]] = None,
162 literal_execute: bool = False,
163) -> BindParameter[Any]: ...
166def literal(
167 value: Any,
168 type_: Optional[_CovariantTypeEngineArgument[Any]] = None,
169 literal_execute: bool = False,
170) -> BindParameter[Any]:
171 r"""Return a literal clause, bound to a bind parameter.
173 Literal clauses are created automatically when non-
174 :class:`_expression.ClauseElement` objects (such as strings, ints, dates,
175 etc.) are
176 used in a comparison operation with a :class:`_expression.ColumnElement`
177 subclass,
178 such as a :class:`~sqlalchemy.schema.Column` object. Use this function
179 to force the generation of a literal clause, which will be created as a
180 :class:`BindParameter` with a bound value.
182 :param value: the value to be bound. Can be any Python object supported by
183 the underlying DB-API, or is translatable via the given type argument.
185 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine` which will
186 provide bind-parameter translation for this literal.
188 :param literal_execute: optional bool, when True, the SQL engine will
189 attempt to render the bound value directly in the SQL statement at
190 execution time rather than providing as a parameter value.
192 .. versionadded:: 2.0
194 """
195 return coercions.expect(
196 roles.LiteralValueRole,
197 value,
198 type_=type_,
199 literal_execute=literal_execute,
200 )
203def literal_column(
204 text: str, type_: Optional[_CovariantTypeEngineArgument[_T]] = None
205) -> ColumnClause[_T]:
206 r"""Produce a :class:`.ColumnClause` object that has the
207 :paramref:`_expression.column.is_literal` flag set to True.
209 :func:`_expression.literal_column` is similar to
210 :func:`_expression.column`, except that
211 it is more often used as a "standalone" column expression that renders
212 exactly as stated; while :func:`_expression.column`
213 stores a string name that
214 will be assumed to be part of a table and may be quoted as such,
215 :func:`_expression.literal_column` can be that,
216 or any other arbitrary column-oriented
217 expression.
219 :param text: the text of the expression; can be any SQL expression.
220 Quoting rules will not be applied. To specify a column-name expression
221 which should be subject to quoting rules, use the :func:`column`
222 function.
224 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine`
225 object which will
226 provide result-set translation and additional expression semantics for
227 this column. If left as ``None`` the type will be :class:`.NullType`.
229 .. seealso::
231 :func:`_expression.column`
233 :func:`_expression.text`
235 :ref:`tutorial_select_arbitrary_text`
237 """
238 return ColumnClause(text, type_=type_, is_literal=True)
241class CompilerElement(Visitable):
242 """base class for SQL elements that can be compiled to produce a
243 SQL string.
245 .. versionadded:: 2.0
247 """
249 __slots__ = ()
250 __visit_name__ = "compiler_element"
252 supports_execution = False
254 stringify_dialect = "default"
256 @util.preload_module("sqlalchemy.engine.default")
257 @util.preload_module("sqlalchemy.engine.url")
258 def compile(
259 self,
260 bind: Optional[_HasDialect] = None,
261 dialect: Optional[Dialect] = None,
262 **kw: Any,
263 ) -> Compiled:
264 """Compile this SQL expression.
266 The return value is a :class:`~.Compiled` object.
267 Calling ``str()`` or ``unicode()`` on the returned value will yield a
268 string representation of the result. The
269 :class:`~.Compiled` object also can return a
270 dictionary of bind parameter names and values
271 using the ``params`` accessor.
273 :param bind: An :class:`.Connection` or :class:`.Engine` which
274 can provide a :class:`.Dialect` in order to generate a
275 :class:`.Compiled` object. If the ``bind`` and
276 ``dialect`` parameters are both omitted, a default SQL compiler
277 is used.
279 :param column_keys: Used for INSERT and UPDATE statements, a list of
280 column names which should be present in the VALUES clause of the
281 compiled statement. If ``None``, all columns from the target table
282 object are rendered.
284 :param dialect: A :class:`.Dialect` instance which can generate
285 a :class:`.Compiled` object. This argument takes precedence over
286 the ``bind`` argument.
288 :param compile_kwargs: optional dictionary of additional parameters
289 that will be passed through to the compiler within all "visit"
290 methods. This allows any custom flag to be passed through to
291 a custom compilation construct, for example. It is also used
292 for the case of passing the ``literal_binds`` flag through::
294 from sqlalchemy.sql import table, column, select
296 t = table("t", column("x"))
298 s = select(t).where(t.c.x == 5)
300 print(s.compile(compile_kwargs={"literal_binds": True}))
302 .. seealso::
304 :ref:`faq_sql_expression_string`
306 """
308 if dialect is None:
309 if bind:
310 dialect = bind.dialect
311 elif self.stringify_dialect == "default":
312 dialect = self._default_dialect()
313 else:
314 url = util.preloaded.engine_url
315 dialect = url.URL.create(
316 self.stringify_dialect
317 ).get_dialect()()
319 return self._compiler(dialect, **kw)
321 def _default_dialect(self):
322 default = util.preloaded.engine_default
323 return default.StrCompileDialect()
325 def _compiler(self, dialect: Dialect, **kw: Any) -> Compiled:
326 """Return a compiler appropriate for this ClauseElement, given a
327 Dialect."""
329 if TYPE_CHECKING:
330 assert isinstance(self, ClauseElement)
331 return dialect.statement_compiler(dialect, self, **kw)
333 def __str__(self) -> str:
334 return str(self.compile())
337@inspection._self_inspects
338class ClauseElement(
339 SupportsWrappingAnnotations,
340 MemoizedHasCacheKey,
341 HasCopyInternals,
342 ExternallyTraversible,
343 CompilerElement,
344):
345 """Base class for elements of a programmatically constructed SQL
346 expression.
348 """
350 __visit_name__ = "clause"
352 if TYPE_CHECKING:
354 @util.memoized_property
355 def _propagate_attrs(self) -> _PropagateAttrsType:
356 """like annotations, however these propagate outwards liberally
357 as SQL constructs are built, and are set up at construction time.
359 """
360 ...
362 else:
363 _propagate_attrs = util.EMPTY_DICT
365 @util.ro_memoized_property
366 def description(self) -> Optional[str]:
367 return None
369 _is_clone_of: Optional[Self] = None
371 is_clause_element = True
372 is_selectable = False
373 is_dml = False
374 _is_column_element = False
375 _is_keyed_column_element = False
376 _is_table = False
377 _gen_static_annotations_cache_key = False
378 _is_textual = False
379 _is_from_clause = False
380 _is_returns_rows = False
381 _is_text_clause = False
382 _is_from_container = False
383 _is_select_container = False
384 _is_select_base = False
385 _is_select_statement = False
386 _is_bind_parameter = False
387 _is_clause_list = False
388 _is_lambda_element = False
389 _is_singleton_constant = False
390 _is_immutable = False
391 _is_star = False
393 @property
394 def _order_by_label_element(self) -> Optional[Label[Any]]:
395 return None
397 _cache_key_traversal: _CacheKeyTraversalType = None
399 negation_clause: ColumnElement[bool]
401 if typing.TYPE_CHECKING:
403 def get_children(
404 self, *, omit_attrs: typing_Tuple[str, ...] = ..., **kw: Any
405 ) -> Iterable[ClauseElement]: ...
407 @util.ro_non_memoized_property
408 def _from_objects(self) -> List[FromClause]:
409 return []
411 def _set_propagate_attrs(self, values: Mapping[str, Any]) -> Self:
412 # usually, self._propagate_attrs is empty here. one case where it's
413 # not is a subquery against ORM select, that is then pulled as a
414 # property of an aliased class. should all be good
416 # assert not self._propagate_attrs
418 self._propagate_attrs = util.immutabledict(values)
419 return self
421 def _default_compiler(self) -> SQLCompiler:
422 dialect = self._default_dialect()
423 return dialect.statement_compiler(dialect, self) # type: ignore[no-any-return] # noqa: E501
425 def _clone(self, **kw: Any) -> Self:
426 """Create a shallow copy of this ClauseElement.
428 This method may be used by a generative API. Its also used as
429 part of the "deep" copy afforded by a traversal that combines
430 the _copy_internals() method.
432 """
434 skip = self._memoized_keys
435 c = self.__class__.__new__(self.__class__)
437 if skip:
438 # ensure this iteration remains atomic
439 c.__dict__ = {
440 k: v for k, v in self.__dict__.copy().items() if k not in skip
441 }
442 else:
443 c.__dict__ = self.__dict__.copy()
445 # this is a marker that helps to "equate" clauses to each other
446 # when a Select returns its list of FROM clauses. the cloning
447 # process leaves around a lot of remnants of the previous clause
448 # typically in the form of column expressions still attached to the
449 # old table.
450 cc = self._is_clone_of
451 c._is_clone_of = cc if cc is not None else self
452 return c
454 def _negate_in_binary(self, negated_op, original_op):
455 """a hook to allow the right side of a binary expression to respond
456 to a negation of the binary expression.
458 Used for the special case of expanding bind parameter with IN.
460 """
461 return self
463 def _with_binary_element_type(self, type_):
464 """in the context of binary expression, convert the type of this
465 object to the one given.
467 applies only to :class:`_expression.ColumnElement` classes.
469 """
470 return self
472 @property
473 def _constructor(self): # type: ignore[override]
474 """return the 'constructor' for this ClauseElement.
476 This is for the purposes for creating a new object of
477 this type. Usually, its just the element's __class__.
478 However, the "Annotated" version of the object overrides
479 to return the class of its proxied element.
481 """
482 return self.__class__
484 @HasMemoized.memoized_attribute
485 def _cloned_set(self):
486 """Return the set consisting all cloned ancestors of this
487 ClauseElement.
489 Includes this ClauseElement. This accessor tends to be used for
490 FromClause objects to identify 'equivalent' FROM clauses, regardless
491 of transformative operations.
493 """
494 s = util.column_set()
495 f: Optional[ClauseElement] = self
497 # note this creates a cycle, asserted in test_memusage. however,
498 # turning this into a plain @property adds tends of thousands of method
499 # calls to Core / ORM performance tests, so the small overhead
500 # introduced by the relatively small amount of short term cycles
501 # produced here is preferable
502 while f is not None:
503 s.add(f)
504 f = f._is_clone_of
505 return s
507 def _de_clone(self):
508 while self._is_clone_of is not None:
509 self = self._is_clone_of
510 return self
512 @util.ro_non_memoized_property
513 def entity_namespace(self) -> _EntityNamespace:
514 raise AttributeError(
515 "This SQL expression has no entity namespace "
516 "with which to filter from."
517 )
519 def __getstate__(self):
520 d = self.__dict__.copy()
521 d.pop("_is_clone_of", None)
522 d.pop("_generate_cache_key", None)
523 return d
525 def _execute_on_connection(
526 self,
527 connection: Connection,
528 distilled_params: _CoreMultiExecuteParams,
529 execution_options: CoreExecuteOptionsParameter,
530 ) -> Result[Unpack[TupleAny]]:
531 if self.supports_execution:
532 if TYPE_CHECKING:
533 assert isinstance(self, Executable)
534 return connection._execute_clauseelement(
535 self, distilled_params, execution_options
536 )
537 else:
538 raise exc.ObjectNotExecutableError(self)
540 def _execute_on_scalar(
541 self,
542 connection: Connection,
543 distilled_params: _CoreMultiExecuteParams,
544 execution_options: CoreExecuteOptionsParameter,
545 ) -> Any:
546 """an additional hook for subclasses to provide a different
547 implementation for connection.scalar() vs. connection.execute().
549 .. versionadded:: 2.0
551 """
552 return self._execute_on_connection(
553 connection, distilled_params, execution_options
554 ).scalar()
556 def _get_embedded_bindparams(self) -> Sequence[BindParameter[Any]]:
557 """Return the list of :class:`.BindParameter` objects embedded in the
558 object.
560 This accomplishes the same purpose as ``visitors.traverse()`` or
561 similar would provide, however by making use of the cache key
562 it takes advantage of memoization of the key to result in fewer
563 net method calls, assuming the statement is also going to be
564 executed.
566 """
568 key = self._generate_cache_key()
569 if key is None:
570 bindparams: List[BindParameter[Any]] = []
572 traverse(self, {}, {"bindparam": bindparams.append})
573 return bindparams
575 else:
576 return key.bindparams
578 def unique_params(
579 self,
580 __optionaldict: Optional[Dict[str, Any]] = None,
581 /,
582 **kwargs: Any,
583 ) -> Self:
584 """Return a copy with :func:`_expression.bindparam` elements
585 replaced.
587 Same functionality as :meth:`_expression.ClauseElement.params`,
588 except adds `unique=True`
589 to affected bind parameters so that multiple statements can be
590 used.
592 """
593 return self._replace_params(True, __optionaldict, kwargs)
595 def params(
596 self,
597 __optionaldict: Optional[Mapping[str, Any]] = None,
598 /,
599 **kwargs: Any,
600 ) -> Self:
601 """Return a copy with :func:`_expression.bindparam` elements
602 replaced.
604 Returns a copy of this ClauseElement with
605 :func:`_expression.bindparam`
606 elements replaced with values taken from the given dictionary::
608 >>> clause = column("x") + bindparam("foo")
609 >>> print(clause.compile().params)
610 {'foo':None}
611 >>> print(clause.params({"foo": 7}).compile().params)
612 {'foo':7}
614 """
615 return self._replace_params(False, __optionaldict, kwargs)
617 @deprecated(
618 "2.1",
619 "The params() and unique_params() methods on non-statement "
620 "ClauseElement objects are deprecated; params() is now limited to "
621 "statement level objects such as select(), insert(), union(), etc. ",
622 )
623 def _replace_params(
624 self,
625 unique: bool,
626 optionaldict: Optional[Mapping[str, Any]],
627 kwargs: Dict[str, Any],
628 ) -> Self:
629 if optionaldict:
630 kwargs.update(optionaldict)
632 def visit_bindparam(bind: BindParameter[Any]) -> None:
633 if bind.key in kwargs:
634 bind.value = kwargs[bind.key]
635 bind.required = False
636 if unique:
637 bind._convert_to_unique()
639 return cloned_traverse(
640 self,
641 {"maintain_key": True, "detect_subquery_cols": True},
642 {"bindparam": visit_bindparam},
643 )
645 def compare(self, other: ClauseElement, **kw: Any) -> bool:
646 r"""Compare this :class:`_expression.ClauseElement` to
647 the given :class:`_expression.ClauseElement`.
649 Subclasses should override the default behavior, which is a
650 straight identity comparison.
652 \**kw are arguments consumed by subclass ``compare()`` methods and
653 may be used to modify the criteria for comparison
654 (see :class:`_expression.ColumnElement`).
656 """
657 return traversals.compare(self, other, **kw)
659 def self_group(
660 self, against: Optional[OperatorType] = None
661 ) -> ClauseElement:
662 """Apply a 'grouping' to this :class:`_expression.ClauseElement`.
664 This method is overridden by subclasses to return a "grouping"
665 construct, i.e. parenthesis. In particular it's used by "binary"
666 expressions to provide a grouping around themselves when placed into a
667 larger expression, as well as by :func:`_expression.select`
668 constructs when placed into the FROM clause of another
669 :func:`_expression.select`. (Note that subqueries should be
670 normally created using the :meth:`_expression.Select.alias` method,
671 as many
672 platforms require nested SELECT statements to be named).
674 As expressions are composed together, the application of
675 :meth:`self_group` is automatic - end-user code should never
676 need to use this method directly. Note that SQLAlchemy's
677 clause constructs take operator precedence into account -
678 so parenthesis might not be needed, for example, in
679 an expression like ``x OR (y AND z)`` - AND takes precedence
680 over OR.
682 The base :meth:`self_group` method of
683 :class:`_expression.ClauseElement`
684 just returns self.
685 """
686 return self
688 def _ungroup(self) -> ClauseElement:
689 """Return this :class:`_expression.ClauseElement`
690 without any groupings.
691 """
693 return self
695 def _compile_w_cache(
696 self,
697 dialect: Dialect,
698 *,
699 compiled_cache: Optional[CompiledCacheType],
700 column_keys: List[str],
701 for_executemany: bool = False,
702 schema_translate_map: Optional[SchemaTranslateMapType] = None,
703 **kw: Any,
704 ) -> tuple[
705 Compiled,
706 Sequence[BindParameter[Any]] | None,
707 _CoreSingleExecuteParams | None,
708 CacheStats,
709 ]:
710 elem_cache_key: Optional[CacheKey]
712 if compiled_cache is not None and dialect._supports_statement_cache:
713 elem_cache_key = self._generate_cache_key()
714 else:
715 elem_cache_key = None
717 extracted_params: Optional[Sequence[BindParameter[Any]]]
718 if elem_cache_key is not None:
719 if TYPE_CHECKING:
720 assert compiled_cache is not None
722 cache_key, extracted_params, param_dict = elem_cache_key
723 key = (
724 dialect,
725 cache_key,
726 tuple(column_keys),
727 bool(schema_translate_map),
728 for_executemany,
729 )
730 compiled_sql = compiled_cache.get(key)
732 if compiled_sql is None:
733 cache_hit = dialect.CACHE_MISS
734 compiled_sql = self._compiler(
735 dialect,
736 cache_key=elem_cache_key,
737 column_keys=column_keys,
738 for_executemany=for_executemany,
739 schema_translate_map=schema_translate_map,
740 **kw,
741 )
742 # ensure that params of the current statement are not
743 # left in the cache
744 assert not compiled_sql._collect_params # type: ignore[attr-defined] # noqa: E501
745 compiled_cache[key] = compiled_sql
746 else:
747 cache_hit = dialect.CACHE_HIT
748 else:
749 param_dict = None
750 extracted_params = None
751 compiled_sql = self._compiler(
752 dialect,
753 cache_key=None,
754 column_keys=column_keys,
755 for_executemany=for_executemany,
756 schema_translate_map=schema_translate_map,
757 **kw,
758 )
759 # here instead the params need to be extracted, since we don't
760 # have them otherwise
761 assert compiled_sql._collect_params # type: ignore[attr-defined] # noqa: E501
763 if not dialect._supports_statement_cache:
764 cache_hit = dialect.NO_DIALECT_SUPPORT
765 elif compiled_cache is None:
766 cache_hit = dialect.CACHING_DISABLED
767 else:
768 cache_hit = dialect.NO_CACHE_KEY
770 return compiled_sql, extracted_params, param_dict, cache_hit
772 def __invert__(self):
773 # undocumented element currently used by the ORM for
774 # relationship.contains()
775 if hasattr(self, "negation_clause"):
776 return self.negation_clause
777 else:
778 return self._negate()
780 def _negate(self) -> ClauseElement:
781 # TODO: this code is uncovered and in all likelihood is not included
782 # in any codepath. So this should raise NotImplementedError in 2.1
783 grouped = self.self_group(against=operators.inv)
784 assert isinstance(grouped, ColumnElement)
785 return UnaryExpression(grouped, operator=operators.inv)
787 def __bool__(self):
788 raise TypeError("Boolean value of this clause is not defined")
790 def __repr__(self):
791 friendly = self.description
792 if friendly is None:
793 return object.__repr__(self)
794 else:
795 return "<%s.%s at 0x%x; %s>" % (
796 self.__module__,
797 self.__class__.__name__,
798 id(self),
799 friendly,
800 )
803class DQLDMLClauseElement(ClauseElement):
804 """represents a :class:`.ClauseElement` that compiles to a DQL or DML
805 expression, not DDL.
807 .. versionadded:: 2.0
809 """
811 if typing.TYPE_CHECKING:
813 def _compiler(self, dialect: Dialect, **kw: Any) -> SQLCompiler:
814 """Return a compiler appropriate for this ClauseElement, given a
815 Dialect."""
816 ...
818 def compile( # noqa: A001
819 self,
820 bind: Optional[_HasDialect] = None,
821 dialect: Optional[Dialect] = None,
822 **kw: Any,
823 ) -> SQLCompiler: ...
826class CompilerColumnElement(
827 roles.DMLColumnRole,
828 roles.DDLConstraintColumnRole,
829 roles.ColumnsClauseRole,
830 CompilerElement,
831):
832 """A compiler-only column element used for ad-hoc string compilations.
834 .. versionadded:: 2.0
836 """
838 __slots__ = ()
840 _propagate_attrs = util.EMPTY_DICT
841 _is_collection_aggregate = False
842 _is_implicitly_boolean = False
844 def _with_binary_element_type(self, type_):
845 raise NotImplementedError()
847 def _gen_cache_key(self, anon_map, bindparams):
848 raise NotImplementedError()
850 @property
851 def _from_objects(self) -> List[FromClause]:
852 raise NotImplementedError()
855# SQLCoreOperations should be suiting the ExpressionElementRole
856# and ColumnsClauseRole. however the MRO issues become too elaborate
857# at the moment.
858class SQLCoreOperations(Generic[_T_co], ColumnOperators, TypingOnly):
859 __slots__ = ()
861 # annotations for comparison methods
862 # these are from operators->Operators / ColumnOperators,
863 # redefined with the specific types returned by ColumnElement hierarchies
864 if typing.TYPE_CHECKING:
866 @util.non_memoized_property
867 def _propagate_attrs(self) -> _PropagateAttrsType: ...
869 def operate(
870 self, op: OperatorType, *other: Any, **kwargs: Any
871 ) -> ColumnElement[Any]: ...
873 def reverse_operate(
874 self, op: OperatorType, other: Any, **kwargs: Any
875 ) -> ColumnElement[Any]: ...
877 @overload
878 def op(
879 self,
880 opstring: str,
881 precedence: int = ...,
882 is_comparison: bool = ...,
883 *,
884 return_type: _TypeEngineArgument[_OPT],
885 python_impl: Optional[Callable[..., Any]] = None,
886 operator_class: OperatorClass = ...,
887 visit_name: Optional[str] = ...,
888 ) -> Callable[[Any], BinaryExpression[_OPT]]: ...
890 @overload
891 def op(
892 self,
893 opstring: str,
894 precedence: int = ...,
895 is_comparison: bool = ...,
896 return_type: Optional[_TypeEngineArgument[Any]] = ...,
897 python_impl: Optional[Callable[..., Any]] = ...,
898 operator_class: OperatorClass = ...,
899 visit_name: Optional[str] = ...,
900 ) -> Callable[[Any], BinaryExpression[Any]]: ...
902 def op(
903 self,
904 opstring: str,
905 precedence: int = 0,
906 is_comparison: bool = False,
907 return_type: Optional[_TypeEngineArgument[Any]] = None,
908 python_impl: Optional[Callable[..., Any]] = None,
909 operator_class: OperatorClass = OperatorClass.BASE,
910 visit_name: Optional[str] = None,
911 ) -> Callable[[Any], BinaryExpression[Any]]: ...
913 def bool_op(
914 self,
915 opstring: str,
916 precedence: int = 0,
917 python_impl: Optional[Callable[..., Any]] = None,
918 ) -> Callable[[Any], BinaryExpression[bool]]: ...
920 def __and__(self, other: Any) -> BooleanClauseList: ...
922 def __or__(self, other: Any) -> BooleanClauseList: ...
924 def __invert__(self) -> ColumnElement[_T_co]: ...
926 def __lt__(self, other: Any) -> ColumnElement[bool]: ...
928 def __le__(self, other: Any) -> ColumnElement[bool]: ...
930 # declare also that this class has an hash method otherwise
931 # it may be assumed to be None by type checkers since the
932 # object defines __eq__ and python sets it to None in that case:
933 # https://docs.python.org/3/reference/datamodel.html#object.__hash__
934 def __hash__(self) -> int: ...
936 def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
937 ...
939 def __ne__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
940 ...
942 def is_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
944 def is_not_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
946 def __gt__(self, other: Any) -> ColumnElement[bool]: ...
948 def __ge__(self, other: Any) -> ColumnElement[bool]: ...
950 def __neg__(self) -> UnaryExpression[_T_co]: ...
952 def __contains__(self, other: Any) -> ColumnElement[bool]: ...
954 def __getitem__(self, index: Any) -> ColumnElement[Any]: ...
956 @overload
957 def __lshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
959 @overload
960 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
962 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
964 @overload
965 def __rlshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
967 @overload
968 def __rlshift__(self, other: Any) -> ColumnElement[Any]: ...
970 def __rlshift__(self, other: Any) -> ColumnElement[Any]: ...
972 @overload
973 def __rshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
975 @overload
976 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
978 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
980 @overload
981 def __rrshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
983 @overload
984 def __rrshift__(self, other: Any) -> ColumnElement[Any]: ...
986 def __rrshift__(self, other: Any) -> ColumnElement[Any]: ...
988 def __matmul__(self, other: Any) -> ColumnElement[Any]: ...
990 def __rmatmul__(self, other: Any) -> ColumnElement[Any]: ...
992 @overload
993 def concat(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
995 @overload
996 def concat(self, other: Any) -> ColumnElement[Any]: ...
998 def concat(self, other: Any) -> ColumnElement[Any]: ...
1000 def like(
1001 self, other: Any, escape: Optional[str] = None
1002 ) -> BinaryExpression[bool]: ...
1004 def ilike(
1005 self, other: Any, escape: Optional[str] = None
1006 ) -> BinaryExpression[bool]: ...
1008 def bitwise_xor(self, other: Any) -> BinaryExpression[Any]: ...
1010 def bitwise_or(self, other: Any) -> BinaryExpression[Any]: ...
1012 def bitwise_and(self, other: Any) -> BinaryExpression[Any]: ...
1014 def bitwise_not(self) -> UnaryExpression[_T_co]: ...
1016 def bitwise_lshift(self, other: Any) -> BinaryExpression[Any]: ...
1018 def bitwise_rshift(self, other: Any) -> BinaryExpression[Any]: ...
1020 def in_(
1021 self,
1022 other: Union[
1023 Iterable[Any], BindParameter[Any], roles.InElementRole
1024 ],
1025 ) -> BinaryExpression[bool]: ...
1027 def not_in(
1028 self,
1029 other: Union[
1030 Iterable[Any], BindParameter[Any], roles.InElementRole
1031 ],
1032 ) -> BinaryExpression[bool]: ...
1034 def notin_(
1035 self,
1036 other: Union[
1037 Iterable[Any], BindParameter[Any], roles.InElementRole
1038 ],
1039 ) -> BinaryExpression[bool]: ...
1041 def not_like(
1042 self, other: Any, escape: Optional[str] = None
1043 ) -> BinaryExpression[bool]: ...
1045 def notlike(
1046 self, other: Any, escape: Optional[str] = None
1047 ) -> BinaryExpression[bool]: ...
1049 def not_ilike(
1050 self, other: Any, escape: Optional[str] = None
1051 ) -> BinaryExpression[bool]: ...
1053 def notilike(
1054 self, other: Any, escape: Optional[str] = None
1055 ) -> BinaryExpression[bool]: ...
1057 def is_(self, other: Any) -> BinaryExpression[bool]: ...
1059 def is_not(self, other: Any) -> BinaryExpression[bool]: ...
1061 def isnot(self, other: Any) -> BinaryExpression[bool]: ...
1063 def startswith(
1064 self,
1065 other: Any,
1066 escape: Optional[str] = None,
1067 autoescape: bool = False,
1068 ) -> ColumnElement[bool]: ...
1070 def istartswith(
1071 self,
1072 other: Any,
1073 escape: Optional[str] = None,
1074 autoescape: bool = False,
1075 ) -> ColumnElement[bool]: ...
1077 def endswith(
1078 self,
1079 other: Any,
1080 escape: Optional[str] = None,
1081 autoescape: bool = False,
1082 ) -> ColumnElement[bool]: ...
1084 def iendswith(
1085 self,
1086 other: Any,
1087 escape: Optional[str] = None,
1088 autoescape: bool = False,
1089 ) -> ColumnElement[bool]: ...
1091 def contains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1093 def icontains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1095 def match(self, other: Any, **kwargs: Any) -> ColumnElement[bool]: ...
1097 def regexp_match(
1098 self, pattern: Any, flags: Optional[str] = None
1099 ) -> ColumnElement[bool]: ...
1101 def regexp_replace(
1102 self, pattern: Any, replacement: Any, flags: Optional[str] = None
1103 ) -> ColumnElement[str]: ...
1105 def desc(self) -> UnaryExpression[_T_co]: ...
1107 def asc(self) -> UnaryExpression[_T_co]: ...
1109 def nulls_first(self) -> UnaryExpression[_T_co]: ...
1111 def nullsfirst(self) -> UnaryExpression[_T_co]: ...
1113 def nulls_last(self) -> UnaryExpression[_T_co]: ...
1115 def nullslast(self) -> UnaryExpression[_T_co]: ...
1117 def collate(
1118 self, collation: str, collation_schema: Optional[str] = None
1119 ) -> CollationClause: ...
1121 def between(
1122 self, cleft: Any, cright: Any, symmetric: bool = False
1123 ) -> BinaryExpression[bool]: ...
1125 def distinct(self: _SQO[_T_co]) -> UnaryExpression[_T_co]: ...
1127 def any_(self) -> CollectionAggregate[Any]: ...
1129 def all_(self) -> CollectionAggregate[Any]: ...
1131 # numeric overloads. These need more tweaking
1132 # in particular they all need to have a variant for Optional[_T]
1133 # because Optional only applies to the data side, not the expression
1134 # side
1136 @overload
1137 def __add__(
1138 self: _SQO[_NMT],
1139 other: Any,
1140 ) -> ColumnElement[_NMT]: ...
1142 @overload
1143 def __add__(
1144 self: _SQO[str],
1145 other: Any,
1146 ) -> ColumnElement[str]: ...
1148 @overload
1149 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1151 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1153 @overload
1154 def __radd__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1156 @overload
1157 def __radd__(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
1159 def __radd__(self, other: Any) -> ColumnElement[Any]: ...
1161 @overload
1162 def __sub__(
1163 self: _SQO[_NMT],
1164 other: Any,
1165 ) -> ColumnElement[_NMT]: ...
1167 @overload
1168 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1170 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1172 @overload
1173 def __rsub__(
1174 self: _SQO[_NMT],
1175 other: Any,
1176 ) -> ColumnElement[_NMT]: ...
1178 @overload
1179 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1181 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1183 @overload
1184 def __mul__(
1185 self: _SQO[_NMT],
1186 other: Any,
1187 ) -> ColumnElement[_NMT]: ...
1189 @overload
1190 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1192 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1194 @overload
1195 def __rmul__(
1196 self: _SQO[_NMT],
1197 other: Any,
1198 ) -> ColumnElement[_NMT]: ...
1200 @overload
1201 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1203 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1205 @overload
1206 def __mod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1208 @overload
1209 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1211 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1213 @overload
1214 def __rmod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1216 @overload
1217 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1219 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1221 @overload
1222 def __truediv__(
1223 self: _SQO[int], other: Any
1224 ) -> ColumnElement[_NUMERIC]: ...
1226 @overload
1227 def __truediv__(self: _SQO[_NT], other: Any) -> ColumnElement[_NT]: ...
1229 @overload
1230 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1232 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1234 @overload
1235 def __rtruediv__(
1236 self: _SQO[_NMT], other: Any
1237 ) -> ColumnElement[_NUMERIC]: ...
1239 @overload
1240 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1242 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1244 @overload
1245 def __floordiv__(
1246 self: _SQO[_NMT], other: Any
1247 ) -> ColumnElement[_NMT]: ...
1249 @overload
1250 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1252 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1254 @overload
1255 def __rfloordiv__(
1256 self: _SQO[_NMT], other: Any
1257 ) -> ColumnElement[_NMT]: ...
1259 @overload
1260 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1262 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1265class SQLColumnExpression(
1266 SQLCoreOperations[_T_co], roles.ExpressionElementRole[_T_co], TypingOnly
1267):
1268 """A type that may be used to indicate any SQL column element or object
1269 that acts in place of one.
1271 :class:`.SQLColumnExpression` is a base of
1272 :class:`.ColumnElement`, as well as within the bases of ORM elements
1273 such as :class:`.InstrumentedAttribute`, and may be used in :pep:`484`
1274 typing to indicate arguments or return values that should behave
1275 as column expressions.
1277 .. versionadded:: 2.0.0b4
1280 """
1282 __slots__ = ()
1285_SQO = SQLCoreOperations
1288class ColumnElement(
1289 roles.ColumnArgumentOrKeyRole,
1290 roles.StatementOptionRole,
1291 roles.WhereHavingRole,
1292 roles.BinaryElementRole[_T],
1293 roles.OrderByRole,
1294 roles.ColumnsClauseRole,
1295 roles.LimitOffsetRole,
1296 roles.DMLColumnRole,
1297 roles.DDLConstraintColumnRole,
1298 roles.DDLExpressionRole,
1299 SQLColumnExpression[_T],
1300 DQLDMLClauseElement,
1301):
1302 """Represent a column-oriented SQL expression suitable for usage in the
1303 "columns" clause, WHERE clause etc. of a statement.
1305 While the most familiar kind of :class:`_expression.ColumnElement` is the
1306 :class:`_schema.Column` object, :class:`_expression.ColumnElement`
1307 serves as the basis
1308 for any unit that may be present in a SQL expression, including
1309 the expressions themselves, SQL functions, bound parameters,
1310 literal expressions, keywords such as ``NULL``, etc.
1311 :class:`_expression.ColumnElement`
1312 is the ultimate base class for all such elements.
1314 A wide variety of SQLAlchemy Core functions work at the SQL expression
1315 level, and are intended to accept instances of
1316 :class:`_expression.ColumnElement` as
1317 arguments. These functions will typically document that they accept a
1318 "SQL expression" as an argument. What this means in terms of SQLAlchemy
1319 usually refers to an input which is either already in the form of a
1320 :class:`_expression.ColumnElement` object,
1321 or a value which can be **coerced** into
1322 one. The coercion rules followed by most, but not all, SQLAlchemy Core
1323 functions with regards to SQL expressions are as follows:
1325 * a literal Python value, such as a string, integer or floating
1326 point value, boolean, datetime, ``Decimal`` object, or virtually
1327 any other Python object, will be coerced into a "literal bound
1328 value". This generally means that a :func:`.bindparam` will be
1329 produced featuring the given value embedded into the construct; the
1330 resulting :class:`.BindParameter` object is an instance of
1331 :class:`_expression.ColumnElement`.
1332 The Python value will ultimately be sent
1333 to the DBAPI at execution time as a parameterized argument to the
1334 ``execute()`` or ``executemany()`` methods, after SQLAlchemy
1335 type-specific converters (e.g. those provided by any associated
1336 :class:`.TypeEngine` objects) are applied to the value.
1338 * any special object value, typically ORM-level constructs, which
1339 feature an accessor called ``__clause_element__()``. The Core
1340 expression system looks for this method when an object of otherwise
1341 unknown type is passed to a function that is looking to coerce the
1342 argument into a :class:`_expression.ColumnElement` and sometimes a
1343 :class:`_expression.SelectBase` expression.
1344 It is used within the ORM to
1345 convert from ORM-specific objects like mapped classes and
1346 mapped attributes into Core expression objects.
1348 * The Python ``None`` value is typically interpreted as ``NULL``,
1349 which in SQLAlchemy Core produces an instance of :func:`.null`.
1351 A :class:`_expression.ColumnElement` provides the ability to generate new
1352 :class:`_expression.ColumnElement`
1353 objects using Python expressions. This means that Python operators
1354 such as ``==``, ``!=`` and ``<`` are overloaded to mimic SQL operations,
1355 and allow the instantiation of further :class:`_expression.ColumnElement`
1356 instances
1357 which are composed from other, more fundamental
1358 :class:`_expression.ColumnElement`
1359 objects. For example, two :class:`.ColumnClause` objects can be added
1360 together with the addition operator ``+`` to produce
1361 a :class:`.BinaryExpression`.
1362 Both :class:`.ColumnClause` and :class:`.BinaryExpression` are subclasses
1363 of :class:`_expression.ColumnElement`:
1365 .. sourcecode:: pycon+sql
1367 >>> from sqlalchemy.sql import column
1368 >>> column("a") + column("b")
1369 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
1370 >>> print(column("a") + column("b"))
1371 {printsql}a + b
1373 .. seealso::
1375 :class:`_schema.Column`
1377 :func:`_expression.column`
1379 """
1381 __visit_name__ = "column_element"
1383 primary_key: bool = False
1384 _is_clone_of: Optional[ColumnElement[_T]]
1385 _is_column_element = True
1386 _insert_sentinel: bool = False
1387 _omit_from_statements = False
1388 _is_collection_aggregate = False
1390 foreign_keys: AbstractSet[ForeignKey] = frozenset()
1392 @util.memoized_property
1393 def _proxies(self) -> List[ColumnElement[Any]]:
1394 return []
1396 @util.non_memoized_property
1397 def _tq_label(self) -> Optional[str]:
1398 """The named label that can be used to target
1399 this column in a result set in a "table qualified" context.
1401 This label is almost always the label used when
1402 rendering <expr> AS <label> in a SELECT statement when using
1403 the LABEL_STYLE_TABLENAME_PLUS_COL label style, which is what the
1404 legacy ORM ``Query`` object uses as well.
1406 For a regular Column bound to a Table, this is typically the label
1407 <tablename>_<columnname>. For other constructs, different rules
1408 may apply, such as anonymized labels and others.
1410 .. versionchanged:: 1.4.21 renamed from ``._label``
1412 """
1413 return None
1415 key: Optional[str] = None
1416 """The 'key' that in some circumstances refers to this object in a
1417 Python namespace.
1419 This typically refers to the "key" of the column as present in the
1420 ``.c`` collection of a selectable, e.g. ``sometable.c["somekey"]`` would
1421 return a :class:`_schema.Column` with a ``.key`` of "somekey".
1423 """
1425 @HasMemoized.memoized_attribute
1426 def _tq_key_label(self) -> Optional[str]:
1427 """A label-based version of 'key' that in some circumstances refers
1428 to this object in a Python namespace.
1431 _tq_key_label comes into play when a select() statement is constructed
1432 with apply_labels(); in this case, all Column objects in the ``.c``
1433 collection are rendered as <tablename>_<columnname> in SQL; this is
1434 essentially the value of ._label. But to locate those columns in the
1435 ``.c`` collection, the name is along the lines of <tablename>_<key>;
1436 that's the typical value of .key_label.
1438 .. versionchanged:: 1.4.21 renamed from ``._key_label``
1440 """
1441 return self._proxy_key
1443 @property
1444 def _key_label(self) -> Optional[str]:
1445 """legacy; renamed to _tq_key_label"""
1446 return self._tq_key_label
1448 @property
1449 def _label(self) -> Optional[str]:
1450 """legacy; renamed to _tq_label"""
1451 return self._tq_label
1453 @property
1454 def _non_anon_label(self) -> Optional[str]:
1455 """the 'name' that naturally applies this element when rendered in
1456 SQL.
1458 Concretely, this is the "name" of a column or a label in a
1459 SELECT statement; ``<columnname>`` and ``<labelname>`` below:
1461 .. sourcecode:: sql
1463 SELECT <columnmame> FROM table
1465 SELECT column AS <labelname> FROM table
1467 Above, the two names noted will be what's present in the DBAPI
1468 ``cursor.description`` as the names.
1470 If this attribute returns ``None``, it means that the SQL element as
1471 written does not have a 100% fully predictable "name" that would appear
1472 in the ``cursor.description``. Examples include SQL functions, CAST
1473 functions, etc. While such things do return names in
1474 ``cursor.description``, they are only predictable on a
1475 database-specific basis; e.g. an expression like ``MAX(table.col)`` may
1476 appear as the string ``max`` on one database (like PostgreSQL) or may
1477 appear as the whole expression ``max(table.col)`` on SQLite.
1479 The default implementation looks for a ``.name`` attribute on the
1480 object, as has been the precedent established in SQLAlchemy for many
1481 years. An exception is made on the ``FunctionElement`` subclass
1482 so that the return value is always ``None``.
1484 .. versionadded:: 1.4.21
1488 """
1489 return getattr(self, "name", None)
1491 _render_label_in_columns_clause = True
1492 """A flag used by select._columns_plus_names that helps to determine
1493 we are actually going to render in terms of "SELECT <col> AS <label>".
1494 This flag can be returned as False for some Column objects that want
1495 to be rendered as simple "SELECT <col>"; typically columns that don't have
1496 any parent table and are named the same as what the label would be
1497 in any case.
1499 """
1501 _allow_label_resolve = True
1502 """A flag that can be flipped to prevent a column from being resolvable
1503 by string label name.
1505 The joined eager loader strategy in the ORM uses this, for example.
1507 """
1509 _is_implicitly_boolean = False
1511 _alt_names: Sequence[str] = ()
1513 if TYPE_CHECKING:
1515 def _ungroup(self) -> ColumnElement[_T]: ...
1517 @overload
1518 def self_group(self, against: None = None) -> ColumnElement[_T]: ...
1520 @overload
1521 def self_group(
1522 self, against: Optional[OperatorType] = None
1523 ) -> ColumnElement[Any]: ...
1525 def self_group(
1526 self, against: Optional[OperatorType] = None
1527 ) -> ColumnElement[Any]:
1528 if (
1529 against in (operators.and_, operators.or_, operators._asbool)
1530 and self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity
1531 ):
1532 return AsBoolean(self, operators.is_true, operators.is_false)
1533 elif against in (operators.any_op, operators.all_op):
1534 return Grouping(self)
1535 else:
1536 return self
1538 @overload
1539 def _negate(self: ColumnElement[bool]) -> ColumnElement[bool]: ...
1541 @overload
1542 def _negate(self: ColumnElement[_T]) -> ColumnElement[_T]: ...
1544 def _negate(self) -> ColumnElement[Any]:
1545 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
1546 return AsBoolean(self, operators.is_false, operators.is_true)
1547 else:
1548 grouped = self.self_group(against=operators.inv)
1549 assert isinstance(grouped, ColumnElement)
1550 return UnaryExpression(
1551 grouped,
1552 operator=operators.inv,
1553 )
1555 type: TypeEngine[_T]
1557 if not TYPE_CHECKING:
1559 @util.memoized_property
1560 def type(self) -> TypeEngine[_T]: # noqa: A001
1561 # used for delayed setup of
1562 # type_api
1563 return type_api.NULLTYPE
1565 @HasMemoized.memoized_attribute
1566 def comparator(self) -> TypeEngine.Comparator[_T]:
1567 try:
1568 comparator_factory = self.type.comparator_factory
1569 except AttributeError as err:
1570 raise TypeError(
1571 "Object %r associated with '.type' attribute "
1572 "is not a TypeEngine class or object" % self.type
1573 ) from err
1574 else:
1575 return comparator_factory(self)
1577 def __setstate__(self, state):
1578 self.__dict__.update(state)
1580 def __getattr__(self, key: str) -> Any:
1581 try:
1582 return getattr(self.comparator, key)
1583 except AttributeError as err:
1584 raise AttributeError(
1585 "Neither %r object nor %r object has an attribute %r"
1586 % (
1587 type(self).__name__,
1588 type(self.comparator).__name__,
1589 key,
1590 )
1591 ) from err
1593 def operate(
1594 self,
1595 op: operators.OperatorType,
1596 *other: Any,
1597 **kwargs: Any,
1598 ) -> ColumnElement[Any]:
1599 return op(self.comparator, *other, **kwargs) # type: ignore[no-any-return] # noqa: E501
1601 def reverse_operate(
1602 self, op: operators.OperatorType, other: Any, **kwargs: Any
1603 ) -> ColumnElement[Any]:
1604 return op(other, self.comparator, **kwargs) # type: ignore[no-any-return] # noqa: E501
1606 def _bind_param(
1607 self,
1608 operator: operators.OperatorType,
1609 obj: Any,
1610 type_: Optional[TypeEngine[_T]] = None,
1611 expanding: bool = False,
1612 ) -> BindParameter[_T]:
1613 return BindParameter(
1614 None,
1615 obj,
1616 _compared_to_operator=operator,
1617 type_=type_,
1618 _compared_to_type=self.type,
1619 unique=True,
1620 expanding=expanding,
1621 )
1623 @property
1624 def expression(self) -> ColumnElement[Any]:
1625 """Return a column expression.
1627 Part of the inspection interface; returns self.
1629 """
1630 return self
1632 @property
1633 def _select_iterable(self) -> _SelectIterable:
1634 return (self,)
1636 @util.memoized_property
1637 def base_columns(self) -> FrozenSet[ColumnElement[Any]]:
1638 return frozenset(c for c in self.proxy_set if not c._proxies)
1640 @util.memoized_property
1641 def proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1642 """set of all columns we are proxying
1644 as of 2.0 this is explicitly deannotated columns. previously it was
1645 effectively deannotated columns but wasn't enforced. annotated
1646 columns should basically not go into sets if at all possible because
1647 their hashing behavior is very non-performant.
1649 """
1650 return frozenset([self._deannotate()]).union(
1651 itertools.chain(*[c.proxy_set for c in self._proxies])
1652 )
1654 @util.memoized_property
1655 def _expanded_proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1656 return frozenset(_expand_cloned(self.proxy_set))
1658 def _uncached_proxy_list(self) -> List[ColumnElement[Any]]:
1659 """An 'uncached' version of proxy set.
1661 This list includes annotated columns which perform very poorly in
1662 set operations.
1664 """
1666 return [self] + list(
1667 itertools.chain(*[c._uncached_proxy_list() for c in self._proxies])
1668 )
1670 def shares_lineage(self, othercolumn: ColumnElement[Any]) -> bool:
1671 """Return True if the given :class:`_expression.ColumnElement`
1672 has a common ancestor to this :class:`_expression.ColumnElement`."""
1674 return bool(self.proxy_set.intersection(othercolumn.proxy_set))
1676 def _compare_name_for_result(self, other: ColumnElement[Any]) -> bool:
1677 """Return True if the given column element compares to this one
1678 when targeting within a result row."""
1680 return (
1681 hasattr(other, "name")
1682 and hasattr(self, "name")
1683 and other.name == self.name
1684 )
1686 @HasMemoized.memoized_attribute
1687 def _proxy_key(self) -> Optional[str]:
1688 if self._annotations and "proxy_key" in self._annotations:
1689 return cast(str, self._annotations["proxy_key"])
1691 name = self.key
1692 if not name:
1693 # there's a bit of a seeming contradiction which is that the
1694 # "_non_anon_label" of a column can in fact be an
1695 # "_anonymous_label"; this is when it's on a column that is
1696 # proxying for an anonymous expression in a subquery.
1697 name = self._non_anon_label
1699 if isinstance(name, _anonymous_label):
1700 return None
1701 else:
1702 return name
1704 @HasMemoized.memoized_attribute
1705 def _expression_label(self) -> Optional[str]:
1706 """a suggested label to use in the case that the column has no name,
1707 which should be used if possible as the explicit 'AS <label>'
1708 where this expression would normally have an anon label.
1710 this is essentially mostly what _proxy_key does except it returns
1711 None if the column has a normal name that can be used.
1713 """
1715 if getattr(self, "name", None) is not None:
1716 return None
1717 elif self._annotations and "proxy_key" in self._annotations:
1718 return cast(str, self._annotations["proxy_key"])
1719 else:
1720 return None
1722 def _make_proxy(
1723 self,
1724 selectable: FromClause,
1725 *,
1726 primary_key: ColumnSet,
1727 foreign_keys: Set[KeyedColumnElement[Any]],
1728 name: Optional[str] = None,
1729 key: Optional[str] = None,
1730 name_is_truncatable: bool = False,
1731 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
1732 **kw: Any,
1733 ) -> typing_Tuple[str, ColumnClause[_T]]:
1734 """Create a new :class:`_expression.ColumnElement` representing this
1735 :class:`_expression.ColumnElement` as it appears in the select list of
1736 a descending selectable.
1738 """
1739 if name is None:
1740 name = self._anon_name_label
1741 if key is None:
1742 key = self._proxy_key
1743 else:
1744 key = name
1746 assert key is not None
1748 co: ColumnClause[_T] = ColumnClause(
1749 (
1750 coercions.expect(roles.TruncatedLabelRole, name)
1751 if name_is_truncatable
1752 else name
1753 ),
1754 type_=getattr(self, "type", None),
1755 _selectable=selectable,
1756 )
1758 co._propagate_attrs = selectable._propagate_attrs
1759 if compound_select_cols:
1760 co._proxies = list(compound_select_cols)
1761 else:
1762 co._proxies = [self]
1763 if selectable._is_clone_of is not None:
1764 co._is_clone_of = selectable._is_clone_of.columns.get(key)
1765 return key, co
1767 def cast(self, type_: _CovariantTypeEngineArgument[_OPT]) -> Cast[_OPT]:
1768 """Produce a type cast, i.e. ``CAST(<expression> AS <type>)``.
1770 This is a shortcut to the :func:`_expression.cast` function.
1772 .. seealso::
1774 :ref:`tutorial_casts`
1776 :func:`_expression.cast`
1778 :func:`_expression.type_coerce`
1780 """
1781 return Cast(self, type_)
1783 def label(self, name: Optional[str]) -> Label[_T]:
1784 """Produce a column label, i.e. ``<columnname> AS <name>``.
1786 This is a shortcut to the :func:`_expression.label` function.
1788 If 'name' is ``None``, an anonymous label name will be generated.
1790 """
1791 return Label(name, self, self.type)
1793 def _anon_label(
1794 self, seed: Optional[str], add_hash: Optional[int] = None
1795 ) -> _anonymous_label:
1796 while self._is_clone_of is not None:
1797 self = self._is_clone_of
1799 # as of 1.4 anonymous label for ColumnElement uses hash(), not id(),
1800 # as the identifier, because a column and its annotated version are
1801 # the same thing in a SQL statement
1802 hash_value = hash(self)
1804 if add_hash:
1805 # this path is used for disambiguating anon labels that would
1806 # otherwise be the same name for the same element repeated.
1807 # an additional numeric value is factored in for each label.
1809 # shift hash(self) (which is id(self), typically 8 byte integer)
1810 # 16 bits leftward. fill extra add_hash on right
1811 assert add_hash < (2 << 15)
1812 assert seed
1813 hash_value = (hash_value << 16) | add_hash
1815 # extra underscore is added for labels with extra hash
1816 # values, to isolate the "deduped anon" namespace from the
1817 # regular namespace. eliminates chance of these
1818 # manufactured hash values overlapping with regular ones for some
1819 # undefined python interpreter
1820 seed = seed + "_"
1822 if isinstance(seed, _anonymous_label):
1823 # NOTE: the space after the hash is required
1824 return _anonymous_label(f"{seed}%({hash_value} )s")
1826 return _anonymous_label.safe_construct(hash_value, seed or "anon")
1828 @util.memoized_property
1829 def _anon_name_label(self) -> str:
1830 """Provides a constant 'anonymous label' for this ColumnElement.
1832 This is a label() expression which will be named at compile time.
1833 The same label() is returned each time ``anon_label`` is called so
1834 that expressions can reference ``anon_label`` multiple times,
1835 producing the same label name at compile time.
1837 The compiler uses this function automatically at compile time
1838 for expressions that are known to be 'unnamed' like binary
1839 expressions and function calls.
1841 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1842 public and is renamed to _anon_name_label. anon_name exists
1843 for backwards compat
1845 """
1846 name = getattr(self, "name", None)
1847 return self._anon_label(name)
1849 @util.memoized_property
1850 def _anon_key_label(self) -> _anonymous_label:
1851 """Provides a constant 'anonymous key label' for this ColumnElement.
1853 Compare to ``anon_label``, except that the "key" of the column,
1854 if available, is used to generate the label.
1856 This is used when a deduplicating key is placed into the columns
1857 collection of a selectable.
1859 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1860 public and is renamed to _anon_key_label. anon_key_label exists
1861 for backwards compat
1863 """
1864 return self._anon_label(self._proxy_key)
1866 @property
1867 @util.deprecated(
1868 "1.4",
1869 "The :attr:`_expression.ColumnElement.anon_label` attribute is now "
1870 "private, and the public accessor is deprecated.",
1871 )
1872 def anon_label(self) -> str:
1873 return self._anon_name_label
1875 @property
1876 @util.deprecated(
1877 "1.4",
1878 "The :attr:`_expression.ColumnElement.anon_key_label` attribute is "
1879 "now private, and the public accessor is deprecated.",
1880 )
1881 def anon_key_label(self) -> str:
1882 return self._anon_key_label
1884 def _dedupe_anon_label_idx(self, idx: int) -> str:
1885 """label to apply to a column that is anon labeled, but repeated
1886 in the SELECT, so that we have to make an "extra anon" label that
1887 disambiguates it from the previous appearance.
1889 these labels come out like "foo_bar_id__1" and have double underscores
1890 in them.
1892 """
1893 label = getattr(self, "name", None)
1895 # current convention is that if the element doesn't have a
1896 # ".name" (usually because it is not NamedColumn), we try to
1897 # use a "table qualified" form for the "dedupe anon" label,
1898 # based on the notion that a label like
1899 # "CAST(casttest.v1 AS DECIMAL) AS casttest_v1__1" looks better than
1900 # "CAST(casttest.v1 AS DECIMAL) AS anon__1"
1902 if label is None:
1903 return self._dedupe_anon_tq_label_idx(idx)
1904 else:
1905 return self._anon_label(label, add_hash=idx)
1907 @util.memoized_property
1908 def _anon_tq_label(self) -> _anonymous_label:
1909 return self._anon_label(getattr(self, "_tq_label", None))
1911 @util.memoized_property
1912 def _anon_tq_key_label(self) -> _anonymous_label:
1913 return self._anon_label(getattr(self, "_tq_key_label", None))
1915 def _dedupe_anon_tq_label_idx(self, idx: int) -> _anonymous_label:
1916 label = getattr(self, "_tq_label", None) or "anon"
1918 return self._anon_label(label, add_hash=idx)
1921class KeyedColumnElement(ColumnElement[_T]):
1922 """ColumnElement where ``.key`` is non-None."""
1924 _is_keyed_column_element = True
1926 key: str
1929class WrapsColumnExpression(ColumnElement[_T]):
1930 """Mixin that defines a :class:`_expression.ColumnElement`
1931 as a wrapper with special
1932 labeling behavior for an expression that already has a name.
1934 .. versionadded:: 1.4
1936 .. seealso::
1938 :ref:`change_4449`
1941 """
1943 @property
1944 def wrapped_column_expression(self) -> ColumnElement[_T]:
1945 raise NotImplementedError()
1947 @util.non_memoized_property
1948 def _tq_label(self) -> Optional[str]:
1949 wce = self.wrapped_column_expression
1950 if hasattr(wce, "_tq_label"):
1951 return wce._tq_label
1952 else:
1953 return None
1955 @property
1956 def _label(self) -> Optional[str]:
1957 return self._tq_label
1959 @property
1960 def _non_anon_label(self) -> Optional[str]:
1961 return None
1963 @util.non_memoized_property
1964 def _anon_name_label(self) -> str:
1965 wce = self.wrapped_column_expression
1967 # this logic tries to get the WrappedColumnExpression to render
1968 # with "<expr> AS <name>", where "<name>" is the natural name
1969 # within the expression itself. e.g. "CAST(table.foo) AS foo".
1970 if not wce._is_text_clause:
1971 nal = wce._non_anon_label
1972 if nal:
1973 return nal
1974 elif hasattr(wce, "_anon_name_label"):
1975 return wce._anon_name_label
1976 return super()._anon_name_label
1978 def _dedupe_anon_label_idx(self, idx: int) -> str:
1979 wce = self.wrapped_column_expression
1980 nal = wce._non_anon_label
1981 if nal:
1982 return self._anon_label(nal + "_")
1983 else:
1984 return self._dedupe_anon_tq_label_idx(idx)
1986 @property
1987 def _proxy_key(self):
1988 wce = self.wrapped_column_expression
1990 if not wce._is_text_clause:
1991 return wce._proxy_key
1992 return super()._proxy_key
1995class DMLTargetCopy(roles.InElementRole, KeyedColumnElement[_T]):
1996 """Refer to another column's VALUES or SET expression in an INSERT or
1997 UPDATE statement.
1999 See the public-facing :func:`_sql.from_dml_column` constructor for
2000 background.
2002 .. versionadded:: 2.1
2005 """
2007 def __init__(self, column: _OnlyColumnArgument[_T]):
2008 self.column = coercions.expect(roles.ColumnArgumentRole, column)
2009 self.type = self.column.type
2011 __visit_name__ = "dmltargetcopy"
2013 _traverse_internals: _TraverseInternalsType = [
2014 ("column", InternalTraversal.dp_clauseelement),
2015 ]
2018class BindParameter(roles.InElementRole, KeyedColumnElement[_T]):
2019 r"""Represent a "bound expression".
2021 :class:`.BindParameter` is invoked explicitly using the
2022 :func:`.bindparam` function, as in::
2024 from sqlalchemy import bindparam
2026 stmt = select(users_table).where(
2027 users_table.c.name == bindparam("username")
2028 )
2030 Detailed discussion of how :class:`.BindParameter` is used is
2031 at :func:`.bindparam`.
2033 .. seealso::
2035 :func:`.bindparam`
2037 """
2039 __visit_name__ = "bindparam"
2041 _traverse_internals: _TraverseInternalsType = [
2042 ("key", InternalTraversal.dp_anon_name),
2043 ("type", InternalTraversal.dp_type),
2044 ("callable", InternalTraversal.dp_plain_dict),
2045 ("value", InternalTraversal.dp_plain_obj),
2046 ("literal_execute", InternalTraversal.dp_boolean),
2047 ]
2049 key: str
2050 _anon_map_key: Optional[str] = None
2051 type: TypeEngine[_T]
2052 value: Optional[_T]
2054 _is_crud = False
2055 _is_bind_parameter = True
2057 # bindparam implements its own _gen_cache_key() method however
2058 # we check subclasses for this flag, else no cache key is generated
2059 inherit_cache = True
2061 def __init__(
2062 self,
2063 key: Optional[str],
2064 value: Any = _NoArg.NO_ARG,
2065 type_: Optional[_CovariantTypeEngineArgument[_T]] = None,
2066 unique: bool = False,
2067 required: Union[bool, Literal[_NoArg.NO_ARG]] = _NoArg.NO_ARG,
2068 quote: Optional[bool] = None,
2069 callable_: Optional[Callable[[], Any]] = None,
2070 expanding: bool = False,
2071 isoutparam: bool = False,
2072 literal_execute: bool = False,
2073 _compared_to_operator: Optional[OperatorType] = None,
2074 _compared_to_type: Optional[TypeEngine[Any]] = None,
2075 _is_crud: bool = False,
2076 ):
2077 if required is _NoArg.NO_ARG:
2078 required = value is _NoArg.NO_ARG and callable_ is None
2079 if value is _NoArg.NO_ARG:
2080 value = None
2082 if quote is not None:
2083 key = quoted_name.construct(key, quote)
2085 if unique:
2086 self.key, self._anon_map_key = (
2087 _anonymous_label.safe_construct_with_key(
2088 id(self),
2089 (
2090 key
2091 if key is not None
2092 and not isinstance(key, _anonymous_label)
2093 else "param"
2094 ),
2095 sanitize_key=True,
2096 )
2097 )
2098 elif key:
2099 self.key = key
2100 else:
2101 self.key, self._anon_map_key = (
2102 _anonymous_label.safe_construct_with_key(id(self), "param")
2103 )
2105 # identifying key that won't change across
2106 # clones, used to identify the bind's logical
2107 # identity
2108 self._identifying_key = self.key
2110 # key that was passed in the first place, used to
2111 # generate new keys
2112 self._orig_key = key or "param"
2114 self.unique = unique
2115 self.value = value
2116 self.callable = callable_
2117 self.isoutparam = isoutparam
2118 self.required = required
2120 # indicate an "expanding" parameter; the compiler sets this
2121 # automatically in the compiler _render_in_expr_w_bindparam method
2122 # for an IN expression
2123 self.expanding = expanding
2125 # this is another hint to help w/ expanding and is typically
2126 # set in the compiler _render_in_expr_w_bindparam method for an
2127 # IN expression
2128 self.expand_op = None
2130 self.literal_execute = literal_execute
2131 if _is_crud:
2132 self._is_crud = True
2134 if type_ is None:
2135 if expanding:
2136 if value:
2137 check_value = value[0]
2138 else:
2139 check_value = type_api._NO_VALUE_IN_LIST
2140 else:
2141 check_value = value
2142 if _compared_to_type is not None:
2143 self.type = _compared_to_type.coerce_compared_value(
2144 _compared_to_operator, check_value
2145 )
2146 else:
2147 self.type = type_api._resolve_value_to_type(check_value)
2148 elif isinstance(type_, type):
2149 self.type = type_() # type: ignore[assignment]
2150 elif is_tuple_type(type_): # type: ignore[arg-type]
2151 if value:
2152 if expanding:
2153 check_value = value[0]
2154 else:
2155 check_value = value
2156 cast("BindParameter[TupleAny]", self).type = (
2157 type_._resolve_values_to_types(check_value)
2158 )
2159 else:
2160 cast("BindParameter[TupleAny]", self).type = type_
2161 else:
2162 self.type = type_ # type: ignore[assignment]
2164 def _with_value(self, value, maintain_key=False, required=NO_ARG):
2165 """Return a copy of this :class:`.BindParameter` with the given value
2166 set.
2167 """
2168 cloned = self._clone(maintain_key=maintain_key)
2169 cloned.value = value
2170 cloned.callable = None
2171 cloned.required = required if required is not NO_ARG else self.required
2172 if cloned.type is type_api.NULLTYPE:
2173 cloned.type = type_api._resolve_value_to_type(value)
2174 return cloned
2176 @property
2177 def effective_value(self) -> Optional[_T]:
2178 """Return the value of this bound parameter,
2179 taking into account if the ``callable`` parameter
2180 was set.
2182 The ``callable`` value will be evaluated
2183 and returned if present, else ``value``.
2185 """
2186 if self.callable:
2187 # TODO: set up protocol for bind parameter callable
2188 return self.callable() # type: ignore[no-any-return]
2189 else:
2190 return self.value
2192 def render_literal_execute(self) -> Self:
2193 """Produce a copy of this bound parameter that will enable the
2194 :paramref:`_sql.BindParameter.literal_execute` flag.
2196 The :paramref:`_sql.BindParameter.literal_execute` flag will
2197 have the effect of the parameter rendered in the compiled SQL
2198 string using ``[POSTCOMPILE]`` form, which is a special form that
2199 is converted to be a rendering of the literal value of the parameter
2200 at SQL execution time. The rationale is to support caching
2201 of SQL statement strings that can embed per-statement literal values,
2202 such as LIMIT and OFFSET parameters, in the final SQL string that
2203 is passed to the DBAPI. Dialects in particular may want to use
2204 this method within custom compilation schemes.
2206 .. versionadded:: 1.4.5
2208 .. seealso::
2210 :ref:`engine_thirdparty_caching`
2212 """
2213 c: Self = ClauseElement._clone(self)
2214 c.literal_execute = True
2215 return c
2217 def _negate_in_binary(self, negated_op, original_op):
2218 if self.expand_op is original_op:
2219 bind = self._clone()
2220 bind.expand_op = negated_op
2221 return bind
2222 else:
2223 return self
2225 def _with_binary_element_type(self, type_: TypeEngine[Any]) -> Self:
2226 c: Self = ClauseElement._clone(self)
2227 c.type = type_
2228 return c
2230 def _clone(self, maintain_key: bool = False, **kw: Any) -> Self:
2231 c: Self = ClauseElement._clone(self, **kw)
2232 # ensure all the BindParameter objects stay in cloned set.
2233 # in #7823, we changed "clone" so that a clone only keeps a reference
2234 # to the "original" element, since for column correspondence, that's
2235 # all we need. However, for BindParam, _cloned_set is used by
2236 # the "cache key bind match" lookup, which means if any of those
2237 # interim BindParameter objects became part of a cache key in the
2238 # cache, we need it. So here, make sure all clones keep carrying
2239 # forward.
2240 c._cloned_set.update(self._cloned_set)
2241 if not maintain_key and self.unique:
2242 c.key, c._anon_map_key = _anonymous_label.safe_construct_with_key(
2243 id(c), c._orig_key or "param", sanitize_key=True
2244 )
2245 return c
2247 def _gen_cache_key(self, anon_map, bindparams):
2248 _gen_cache_ok = self.__class__.__dict__.get("inherit_cache", False)
2250 if not _gen_cache_ok:
2251 if anon_map is not None:
2252 anon_map[NO_CACHE] = True
2253 return None
2255 id_, found = anon_map.get_anon(self)
2256 if found:
2257 return (id_, self.__class__)
2259 if bindparams is not None:
2260 bindparams.append(self)
2262 return (
2263 id_,
2264 self.__class__,
2265 self.type._static_cache_key,
2266 (
2267 anon_map[self._anon_map_key]
2268 if self._anon_map_key is not None
2269 else self.key
2270 ),
2271 self.literal_execute,
2272 )
2274 def _convert_to_unique(self):
2275 if not self.unique:
2276 self.unique = True
2277 self.key, self._anon_map_key = (
2278 _anonymous_label.safe_construct_with_key(
2279 id(self), self._orig_key or "param", sanitize_key=True
2280 )
2281 )
2283 def __getstate__(self):
2284 """execute a deferred value for serialization purposes."""
2286 d = self.__dict__.copy()
2287 v = self.value
2288 if self.callable:
2289 v = self.callable()
2290 d["callable"] = None
2291 d["value"] = v
2292 return d
2294 def __setstate__(self, state):
2295 if state.get("unique", False):
2296 anon_and_key = _anonymous_label.safe_construct_with_key(
2297 id(self), state.get("_orig_key", "param"), sanitize_key=True
2298 )
2299 state["key"], state["_anon_map_key"] = anon_and_key
2300 self.__dict__.update(state)
2302 def __repr__(self):
2303 return "%s(%r, %r, type_=%r)" % (
2304 self.__class__.__name__,
2305 self.key,
2306 self.value,
2307 self.type,
2308 )
2311class TypeClause(DQLDMLClauseElement):
2312 """Handle a type keyword in a SQL statement.
2314 Used by the ``Case`` statement.
2316 """
2318 __visit_name__ = "typeclause"
2320 _traverse_internals: _TraverseInternalsType = [
2321 ("type", InternalTraversal.dp_type)
2322 ]
2323 type: TypeEngine[Any]
2325 def __init__(self, type_: TypeEngine[Any]):
2326 self.type = type_
2329class AbstractTextClause(
2330 roles.DDLConstraintColumnRole,
2331 roles.DDLExpressionRole,
2332 roles.StatementOptionRole,
2333 roles.WhereHavingRole,
2334 roles.OrderByRole,
2335 roles.FromClauseRole,
2336 roles.SelectStatementRole,
2337 roles.InElementRole,
2338 Generative,
2339 ExecutableStatement,
2340 DQLDMLClauseElement,
2341 roles.BinaryElementRole[Any],
2342):
2343 """Base class for textual SQL constructs like TextClause and TString."""
2345 __visit_name__: str
2347 _is_text_clause = True
2348 _is_textual = True
2349 _is_implicitly_boolean = False
2350 _render_label_in_columns_clause = False
2351 _omit_from_statements = False
2352 _is_collection_aggregate = False
2354 @property
2355 def _hide_froms(self) -> Iterable[FromClause]:
2356 return ()
2358 def __and__(self, other):
2359 # support use in select.where(), query.filter()
2360 return and_(self, other)
2362 @property
2363 def _select_iterable(self) -> _SelectIterable:
2364 return (self,)
2366 # help in those cases where text/tstring() is
2367 # interpreted in a column expression situation
2368 key: Optional[str] = None
2369 _label: Optional[str] = None
2371 _allow_label_resolve = False
2373 @property
2374 def type(self) -> TypeEngine[Any]:
2375 return type_api.NULLTYPE
2377 @property
2378 def comparator(self):
2379 return self.type.comparator_factory(self) # type: ignore[arg-type]
2381 def self_group(
2382 self, against: Optional[OperatorType] = None
2383 ) -> Union[Self, Grouping[Any]]:
2384 if against is operators.in_op:
2385 return Grouping(self)
2386 else:
2387 return self
2389 def bindparams(
2390 self,
2391 *binds: BindParameter[Any],
2392 **names_to_values: Any,
2393 ) -> Self:
2394 """Establish the values and/or types of bound parameters within
2395 this :class:`_expression.AbstractTextClause` construct.
2397 This is implemented only for :class:`.TextClause` will raise
2398 ``NotImplementedError`` for :class:`.TString`.
2400 """
2401 raise NotImplementedError()
2403 @util.preload_module("sqlalchemy.sql.selectable")
2404 def columns(
2405 self,
2406 *cols: _OnlyColumnArgument[Any],
2407 **types: _TypeEngineArgument[Any],
2408 ) -> TextualSelect:
2409 r"""Turn this :class:`_expression.AbstractTextClause` object into a
2410 :class:`_expression.TextualSelect`
2411 object that serves the same role as a SELECT
2412 statement.
2414 The :class:`_expression.TextualSelect` is part of the
2415 :class:`_expression.SelectBase`
2416 hierarchy and can be embedded into another statement by using the
2417 :meth:`_expression.TextualSelect.subquery` method to produce a
2418 :class:`.Subquery`
2419 object, which can then be SELECTed from.
2421 This function essentially bridges the gap between an entirely
2422 textual SELECT statement and the SQL expression language concept
2423 of a "selectable"::
2425 from sqlalchemy.sql import column, text
2427 stmt = text("SELECT id, name FROM some_table")
2428 stmt = stmt.columns(column("id"), column("name")).subquery("st")
2430 stmt = (
2431 select(mytable)
2432 .select_from(mytable.join(stmt, mytable.c.name == stmt.c.name))
2433 .where(stmt.c.id > 5)
2434 )
2436 Above, we pass a series of :func:`_expression.column` elements to the
2437 :meth:`_expression.AbstractTextClause.columns` method positionally.
2438 These :func:`_expression.column` elements now become first class
2439 elements upon the :attr:`_expression.TextualSelect.selected_columns`
2440 column collection, which then become part of the :attr:`.Subquery.c`
2441 collection after :meth:`_expression.TextualSelect.subquery` is invoked.
2443 The column expressions we pass to
2444 :meth:`_expression.AbstractTextClause.columns` may also be typed; when
2445 we do so, these :class:`.TypeEngine` objects become the effective
2446 return type of the column, so that SQLAlchemy's result-set-processing
2447 systems may be used on the return values. This is often needed for
2448 types such as date or boolean types, as well as for unicode processing
2449 on some dialect configurations::
2451 stmt = text("SELECT id, name, timestamp FROM some_table")
2452 stmt = stmt.columns(
2453 column("id", Integer),
2454 column("name", Unicode),
2455 column("timestamp", DateTime),
2456 )
2458 for id, name, timestamp in connection.execute(stmt):
2459 print(id, name, timestamp)
2461 As a shortcut to the above syntax, keyword arguments referring to
2462 types alone may be used, if only type conversion is needed::
2464 stmt = text("SELECT id, name, timestamp FROM some_table")
2465 stmt = stmt.columns(id=Integer, name=Unicode, timestamp=DateTime)
2467 for id, name, timestamp in connection.execute(stmt):
2468 print(id, name, timestamp)
2470 The positional form of :meth:`_expression.AbstractTextClause.columns`
2471 also provides the unique feature of **positional column targeting**,
2472 which is particularly useful when using the ORM with complex textual
2473 queries. If we specify the columns from our model to
2474 :meth:`_expression.AbstractTextClause.columns`, the result set will
2475 match to those columns positionally, meaning the name or origin of the
2476 column in the textual SQL doesn't matter::
2478 stmt = text(
2479 "SELECT users.id, addresses.id, users.id, "
2480 "users.name, addresses.email_address AS email "
2481 "FROM users JOIN addresses ON users.id=addresses.user_id "
2482 "WHERE users.id = 1"
2483 ).columns(
2484 User.id,
2485 Address.id,
2486 Address.user_id,
2487 User.name,
2488 Address.email_address,
2489 )
2491 query = (
2492 session.query(User)
2493 .from_statement(stmt)
2494 .options(contains_eager(User.addresses))
2495 )
2497 The :meth:`_expression.AbstractTextClause.columns` method provides a
2498 direct route to calling :meth:`_expression.FromClause.subquery` as well
2499 as :meth:`_expression.SelectBase.cte` against a textual SELECT
2500 statement::
2502 stmt = stmt.columns(id=Integer, name=String).cte("st")
2504 stmt = select(sometable).where(sometable.c.id == stmt.c.id)
2506 :param \*cols: A series of :class:`_expression.ColumnElement` objects,
2507 typically
2508 :class:`_schema.Column` objects from a :class:`_schema.Table`
2509 or ORM level
2510 column-mapped attributes, representing a set of columns that this
2511 textual string will SELECT from.
2513 :param \**types: A mapping of string names to :class:`.TypeEngine`
2514 type objects indicating the datatypes to use for names that are
2515 SELECTed from the textual string. Prefer to use the ``*cols``
2516 argument as it also indicates positional ordering.
2518 """
2519 selectable = util.preloaded.sql_selectable
2521 input_cols: List[NamedColumn[Any]] = [
2522 coercions.expect(roles.LabeledColumnExprRole, col) for col in cols
2523 ]
2525 positional_input_cols = [
2526 (
2527 ColumnClause(col.key, types.pop(col.key))
2528 if col.key in types
2529 else col
2530 )
2531 for col in input_cols
2532 ]
2533 keyed_input_cols: List[NamedColumn[Any]] = [
2534 ColumnClause(key, type_) for key, type_ in types.items()
2535 ]
2537 elem = selectable.TextualSelect.__new__(selectable.TextualSelect)
2538 elem._init(
2539 self,
2540 positional_input_cols + keyed_input_cols,
2541 positional=bool(positional_input_cols) and not keyed_input_cols,
2542 )
2543 return elem
2546class TextClause(AbstractTextClause, inspection.Inspectable["TextClause"]):
2547 """Represent a literal SQL text fragment.
2549 E.g.::
2551 from sqlalchemy import text
2553 t = text("SELECT * FROM users")
2554 result = connection.execute(t)
2556 The :class:`_expression.TextClause` construct is produced using the
2557 :func:`_expression.text`
2558 function; see that function for full documentation.
2560 .. seealso::
2562 :func:`_expression.text`
2564 """
2566 __visit_name__ = "textclause"
2568 _traverse_internals: _TraverseInternalsType = [
2569 ("_bindparams", InternalTraversal.dp_string_clauseelement_dict),
2570 ("text", InternalTraversal.dp_string),
2571 ] + ExecutableStatement._executable_traverse_internals
2573 # also used by the compiler to render the text; the two must find
2574 # the same names, else a name is rendered for which no
2575 # BindParameter exists
2576 _bind_params_regex = re.compile(
2577 r"(?<![:\w\x5c]):(\w+)(?![:\w])", re.UNICODE
2578 )
2580 @property
2581 def _is_star(self) -> bool: # type: ignore[override]
2582 return self.text == "*"
2584 def __init__(self, text: str):
2585 self._bindparams: Dict[str, BindParameter[Any]] = {}
2587 def repl(m):
2588 self._bindparams[m.group(1)] = BindParameter(m.group(1))
2589 return ":%s" % m.group(1)
2591 # scan the string and search for bind parameter names, add them
2592 # to the list of bindparams
2593 self.text = self._bind_params_regex.sub(repl, text)
2595 @_generative
2596 def bindparams(
2597 self,
2598 *binds: BindParameter[Any],
2599 **names_to_values: Any,
2600 ) -> Self:
2601 """Establish the values and/or types of bound parameters within
2602 this :class:`_expression.TextClause` construct.
2604 Given a text construct such as::
2606 from sqlalchemy import text
2608 stmt = text(
2609 "SELECT id, name FROM user WHERE name=:name AND timestamp=:timestamp"
2610 )
2612 the :meth:`_expression.TextClause.bindparams`
2613 method can be used to establish
2614 the initial value of ``:name`` and ``:timestamp``,
2615 using simple keyword arguments::
2617 stmt = stmt.bindparams(
2618 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2619 )
2621 Where above, new :class:`.BindParameter` objects
2622 will be generated with the names ``name`` and ``timestamp``, and
2623 values of ``jack`` and ``datetime.datetime(2012, 10, 8, 15, 12, 5)``,
2624 respectively. The types will be
2625 inferred from the values given, in this case :class:`.String` and
2626 :class:`.DateTime`.
2628 When specific typing behavior is needed, the positional ``*binds``
2629 argument can be used in which to specify :func:`.bindparam` constructs
2630 directly. These constructs must include at least the ``key``
2631 argument, then an optional value and type::
2633 from sqlalchemy import bindparam
2635 stmt = stmt.bindparams(
2636 bindparam("name", value="jack", type_=String),
2637 bindparam("timestamp", type_=DateTime),
2638 )
2640 Above, we specified the type of :class:`.DateTime` for the
2641 ``timestamp`` bind, and the type of :class:`.String` for the ``name``
2642 bind. In the case of ``name`` we also set the default value of
2643 ``"jack"``.
2645 Additional bound parameters can be supplied at statement execution
2646 time, e.g.::
2648 result = connection.execute(
2649 stmt, timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2650 )
2652 The :meth:`_expression.TextClause.bindparams`
2653 method can be called repeatedly,
2654 where it will reuse existing :class:`.BindParameter` objects to add
2655 new information. For example, we can call
2656 :meth:`_expression.TextClause.bindparams`
2657 first with typing information, and a
2658 second time with value information, and it will be combined::
2660 stmt = text(
2661 "SELECT id, name FROM user WHERE name=:name "
2662 "AND timestamp=:timestamp"
2663 )
2664 stmt = stmt.bindparams(
2665 bindparam("name", type_=String), bindparam("timestamp", type_=DateTime)
2666 )
2667 stmt = stmt.bindparams(
2668 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2669 )
2671 The :meth:`_expression.TextClause.bindparams`
2672 method also supports the concept of
2673 **unique** bound parameters. These are parameters that are
2674 "uniquified" on name at statement compilation time, so that multiple
2675 :func:`_expression.text`
2676 constructs may be combined together without the names
2677 conflicting. To use this feature, specify the
2678 :paramref:`.BindParameter.unique` flag on each :func:`.bindparam`
2679 object::
2681 stmt1 = text("select id from table where name=:name").bindparams(
2682 bindparam("name", value="name1", unique=True)
2683 )
2684 stmt2 = text("select id from table where name=:name").bindparams(
2685 bindparam("name", value="name2", unique=True)
2686 )
2688 union = union_all(stmt1.columns(column("id")), stmt2.columns(column("id")))
2690 The above statement will render as:
2692 .. sourcecode:: sql
2694 select id from table where name=:name_1
2695 UNION ALL select id from table where name=:name_2
2697 """ # noqa: E501
2698 self._bindparams = new_params = self._bindparams.copy()
2700 for bind in binds:
2701 try:
2702 # the regex used for text() currently will not match
2703 # a unique/anonymous key in any case, so use the _orig_key
2704 # so that a text() construct can support unique parameters
2705 existing = new_params[bind._orig_key]
2706 except KeyError as err:
2707 raise exc.ArgumentError(
2708 "This text() construct doesn't define a "
2709 "bound parameter named %r" % bind._orig_key
2710 ) from err
2711 else:
2712 new_params[existing._orig_key] = bind
2714 for key, value in names_to_values.items():
2715 try:
2716 existing = new_params[key]
2717 except KeyError as err:
2718 raise exc.ArgumentError(
2719 "This text() construct doesn't define a "
2720 "bound parameter named %r" % key
2721 ) from err
2722 else:
2723 new_params[key] = existing._with_value(value, required=False)
2724 return self
2726 @property
2727 def type(self) -> TypeEngine[Any]:
2728 return type_api.NULLTYPE
2730 @property
2731 def comparator(self):
2732 # TODO: this seems wrong, it seems like we might not
2733 # be using this method.
2734 return self.type.comparator_factory(self) # type: ignore[arg-type]
2736 def self_group(
2737 self, against: Optional[OperatorType] = None
2738 ) -> Union[Self, Grouping[Any]]:
2739 if against is operators.in_op:
2740 return Grouping(self)
2741 else:
2742 return self
2745class _TStringText(ClauseElement):
2746 """Represent a literal string portion of a :class:`.TString`.
2748 Unlike :class:`.TextClause`, the text is not scanned for bound parameter
2749 names and has no escaping rules; it is rendered as is.
2751 .. versionadded:: 2.1.2
2753 """
2755 __visit_name__ = "tstring_text"
2757 _traverse_internals: _TraverseInternalsType = [
2758 ("text", InternalTraversal.dp_string)
2759 ]
2761 def __init__(self, text: str):
2762 self.text = text
2764 @property
2765 def _is_star(self) -> bool: # type: ignore[override]
2766 return self.text == "*"
2769class TString(AbstractTextClause, inspection.Inspectable["TString"]):
2770 """Represent a SQL template string using Python 3.14+ t-strings.
2772 E.g.::
2774 from sqlalchemy import tstring, column
2776 a = 5
2777 b = 10
2778 stmt = tstring(t"select {a}, {b}")
2779 result = connection.execute(stmt)
2781 The :class:`_expression.TString` construct is produced using the
2782 :func:`_expression.tstring` function; see that function for full
2783 documentation.
2785 .. versionadded:: 2.1
2787 .. seealso::
2789 :func:`_expression.tstring`
2791 """
2793 __visit_name__ = "tstring"
2795 _traverse_internals: _TraverseInternalsType = [
2796 ("parts", InternalTraversal.dp_clauseelement_list)
2797 ] + ExecutableStatement._executable_traverse_internals
2799 @property
2800 def _is_star(self) -> bool: # type: ignore[override]
2801 return (
2802 len(self.parts) == 1
2803 and isinstance(self.parts[0], _TStringText)
2804 and self.parts[0]._is_star
2805 )
2807 def __init__(self, template: Template):
2808 """Construct a :class:`_expression.TString` from a Python 3.14+
2809 template string.
2811 :param template: a Python 3.14+ template string (t-string) that
2812 contains SQL fragments and Python expressions to be interpolated.
2814 """
2815 self.parts: List[ClauseElement] = []
2817 if not isinstance(template, Template):
2818 raise exc.ArgumentError("pep-750 Tstring (e.g. t'...') expected")
2820 for part in template:
2821 if isinstance(part, str):
2822 self.parts.append(_TStringText(part))
2823 else:
2824 assert hasattr(part, "value")
2825 self.parts.append(
2826 coercions.expect(roles.TStringElementRole, part.value)
2827 )
2829 def bindparams(
2830 self,
2831 *binds: BindParameter[Any],
2832 **names_to_values: Any,
2833 ) -> Self:
2834 """Not supported for TString constructs.
2836 TString constructs do not support .bindparams(). Bind parameters
2837 are automatically created from interpolated values.
2839 """
2840 raise NotImplementedError(
2841 "TString constructs do not support .bindparams(). "
2842 "Bind parameters are automatically created "
2843 "from interpolated values."
2844 )
2847class Null(SingletonConstant, roles.ConstExprRole[None], ColumnElement[None]):
2848 """Represent the NULL keyword in a SQL statement.
2850 :class:`.Null` is accessed as a constant via the
2851 :func:`.null` function.
2853 """
2855 __visit_name__ = "null"
2857 _traverse_internals: _TraverseInternalsType = []
2858 _singleton: Null
2860 if not TYPE_CHECKING:
2862 @util.memoized_property
2863 def type(self) -> TypeEngine[_T]: # noqa: A001
2864 return type_api.NULLTYPE
2866 @classmethod
2867 def _instance(cls) -> Null:
2868 """Return a constant :class:`.Null` construct."""
2870 return Null._singleton
2873Null._create_singleton()
2876class False_(
2877 SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]
2878):
2879 """Represent the ``false`` keyword, or equivalent, in a SQL statement.
2881 :class:`.False_` is accessed as a constant via the
2882 :func:`.false` function.
2884 """
2886 __visit_name__ = "false"
2887 _traverse_internals: _TraverseInternalsType = []
2888 _singleton: False_
2890 if not TYPE_CHECKING:
2892 @util.memoized_property
2893 def type(self) -> TypeEngine[_T]: # noqa: A001
2894 return type_api.BOOLEANTYPE
2896 def _negate(self) -> True_:
2897 return True_._singleton
2899 @classmethod
2900 def _instance(cls) -> False_:
2901 return False_._singleton
2904False_._create_singleton()
2907class True_(SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]):
2908 """Represent the ``true`` keyword, or equivalent, in a SQL statement.
2910 :class:`.True_` is accessed as a constant via the
2911 :func:`.true` function.
2913 """
2915 __visit_name__ = "true"
2917 _traverse_internals: _TraverseInternalsType = []
2918 _singleton: True_
2920 if not TYPE_CHECKING:
2922 @util.memoized_property
2923 def type(self) -> TypeEngine[_T]: # noqa: A001
2924 return type_api.BOOLEANTYPE
2926 def _negate(self) -> False_:
2927 return False_._singleton
2929 @classmethod
2930 def _ifnone(
2931 cls, other: Optional[ColumnElement[Any]]
2932 ) -> ColumnElement[Any]:
2933 if other is None:
2934 return cls._instance()
2935 else:
2936 return other
2938 @classmethod
2939 def _instance(cls) -> True_:
2940 return True_._singleton
2943True_._create_singleton()
2946class ElementList(DQLDMLClauseElement):
2947 """Describe a list of clauses that will be space separated.
2949 This is a minimal version of :class:`.ClauseList` which is used by
2950 the :class:`.HasSyntaxExtension` class. It does not do any coercions
2951 so should be used internally only.
2953 .. versionadded:: 2.1
2955 """
2957 __visit_name__ = "element_list"
2959 _traverse_internals: _TraverseInternalsType = [
2960 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2961 ]
2963 clauses: typing_Tuple[ClauseElement, ...]
2965 def __init__(self, clauses: Sequence[ClauseElement]):
2966 self.clauses = tuple(clauses)
2969class OrderByList(
2970 roles.OrderByRole,
2971 operators.OrderingOperators,
2972 DQLDMLClauseElement,
2973):
2974 """Describe a list of clauses that will be comma separated to nest
2975 within an ORDER BY.
2977 .. versionadded:: 2.1
2979 """
2981 __visit_name__ = "order_by_list"
2983 _traverse_internals: _TraverseInternalsType = [
2984 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2985 ]
2987 clauses: List[ColumnElement[Any]]
2989 def __init__(
2990 self,
2991 clauses: Iterable[Union[OrderByList, _ColumnExpressionArgument[Any]]],
2992 ):
2993 text_converter_role: Type[roles.SQLRole] = roles.ByOfRole
2994 self._text_converter_role = text_converter_role
2996 self.clauses = [
2997 coercions.expect(
2998 text_converter_role, clause, apply_propagate_attrs=self
2999 )
3000 for clause in clauses
3001 ]
3003 def __iter__(self) -> Iterator[ColumnElement[Any]]:
3004 return iter(self.clauses)
3006 def __len__(self) -> int:
3007 return len(self.clauses)
3009 @property
3010 def _select_iterable(self) -> _SelectIterable:
3011 return itertools.chain.from_iterable(
3012 [elem._select_iterable for elem in self.clauses]
3013 )
3015 @util.ro_non_memoized_property
3016 def _from_objects(self) -> List[FromClause]:
3017 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
3019 def self_group(
3020 self, against: Optional[OperatorType] = None
3021 ) -> Union[Self, Grouping[Any]]:
3022 return self
3024 def desc(self) -> OrderByList:
3025 return OrderByList([e.desc() for e in self.clauses])
3027 def asc(self) -> OrderByList:
3028 return OrderByList([e.asc() for e in self.clauses])
3030 def nulls_first(self) -> OrderByList:
3031 return OrderByList([e.nulls_first() for e in self.clauses])
3033 def nulls_last(self) -> OrderByList:
3034 return OrderByList([e.nulls_last() for e in self.clauses])
3037class ClauseList(
3038 roles.InElementRole,
3039 roles.OrderByRole,
3040 roles.ColumnsClauseRole,
3041 roles.DMLColumnRole,
3042 DQLDMLClauseElement,
3043):
3044 """Describe a list of clauses, separated by an operator.
3046 By default, is comma-separated, such as a column listing.
3048 """
3050 __visit_name__ = "clauselist"
3052 # Used by ORM context.py to identify ClauseList objects in legacy
3053 # composite attribute queries (see test_query_cols_legacy test)
3054 _is_clause_list = True
3056 _traverse_internals: _TraverseInternalsType = [
3057 ("clauses", InternalTraversal.dp_clauseelement_list),
3058 ("operator", InternalTraversal.dp_operator),
3059 ]
3061 clauses: List[ColumnElement[Any]]
3063 def __init__(
3064 self,
3065 *clauses: _ColumnExpressionArgument[Any],
3066 operator: OperatorType = operators.comma_op,
3067 group: bool = True,
3068 group_contents: bool = True,
3069 _literal_as_text_role: Type[roles.SQLRole] = roles.WhereHavingRole,
3070 ):
3071 self.operator = operator
3072 self.group = group
3073 self.group_contents = group_contents
3074 clauses_iterator: Iterable[_ColumnExpressionArgument[Any]] = clauses
3075 text_converter_role: Type[roles.SQLRole] = _literal_as_text_role
3076 self._text_converter_role = text_converter_role
3078 if self.group_contents:
3079 self.clauses = [
3080 coercions.expect(
3081 text_converter_role, clause, apply_propagate_attrs=self
3082 ).self_group(against=self.operator)
3083 for clause in clauses_iterator
3084 ]
3085 else:
3086 self.clauses = [
3087 coercions.expect(
3088 text_converter_role, clause, apply_propagate_attrs=self
3089 )
3090 for clause in clauses_iterator
3091 ]
3092 self._is_implicitly_boolean = operators.is_boolean(self.operator)
3094 @classmethod
3095 def _construct_raw(
3096 cls,
3097 operator: OperatorType,
3098 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
3099 ) -> ClauseList:
3100 self = cls.__new__(cls)
3101 self.clauses = list(clauses) if clauses else []
3102 self.group = True
3103 self.operator = operator
3104 self.group_contents = True
3105 self._is_implicitly_boolean = False
3106 return self
3108 def __iter__(self) -> Iterator[ColumnElement[Any]]:
3109 return iter(self.clauses)
3111 def __len__(self) -> int:
3112 return len(self.clauses)
3114 @property
3115 def _select_iterable(self) -> _SelectIterable:
3116 return itertools.chain.from_iterable(
3117 [elem._select_iterable for elem in self.clauses]
3118 )
3120 def append(self, clause):
3121 if self.group_contents:
3122 self.clauses.append(
3123 coercions.expect(self._text_converter_role, clause).self_group(
3124 against=self.operator
3125 )
3126 )
3127 else:
3128 self.clauses.append(
3129 coercions.expect(self._text_converter_role, clause)
3130 )
3132 @util.ro_non_memoized_property
3133 def _from_objects(self) -> List[FromClause]:
3134 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
3136 def self_group(
3137 self, against: Optional[OperatorType] = None
3138 ) -> Union[Self, Grouping[Any]]:
3139 if self.group and operators.is_precedent(self.operator, against):
3140 return Grouping(self)
3141 else:
3142 return self
3145class OperatorExpression(ColumnElement[_T]):
3146 """base for expressions that contain an operator and operands
3148 .. versionadded:: 2.0
3150 """
3152 operator: OperatorType
3153 type: TypeEngine[_T]
3155 group: bool = True
3157 @property
3158 def is_comparison(self):
3159 return operators.is_comparison(self.operator)
3161 def self_group(
3162 self, against: Optional[OperatorType] = None
3163 ) -> Union[Self, Grouping[_T]]:
3164 if (
3165 self.group
3166 and operators.is_precedent(self.operator, against)
3167 or (
3168 # a negate against a non-boolean operator
3169 # doesn't make too much sense but we should
3170 # group for that
3171 against is operators.inv
3172 and not operators.is_boolean(self.operator)
3173 )
3174 ):
3175 return Grouping(self)
3176 else:
3177 return self
3179 @property
3180 def _flattened_operator_clauses(
3181 self,
3182 ) -> typing_Tuple[ColumnElement[Any], ...]:
3183 raise NotImplementedError()
3185 @classmethod
3186 def _construct_for_op(
3187 cls,
3188 left: ColumnElement[Any],
3189 right: ColumnElement[Any],
3190 op: OperatorType,
3191 *,
3192 type_: TypeEngine[_T],
3193 negate: Optional[OperatorType] = None,
3194 modifiers: Optional[Mapping[str, Any]] = None,
3195 ) -> OperatorExpression[_T]:
3196 if operators.is_associative(op):
3197 assert (
3198 negate is None
3199 ), f"negate not supported for associative operator {op}"
3201 multi = False
3202 if getattr(
3203 left, "operator", None
3204 ) is op and type_._compare_type_affinity(left.type):
3205 multi = True
3206 left_flattened = left._flattened_operator_clauses
3207 else:
3208 left_flattened = (left,)
3210 if getattr(
3211 right, "operator", None
3212 ) is op and type_._compare_type_affinity(right.type):
3213 multi = True
3214 right_flattened = right._flattened_operator_clauses
3215 else:
3216 right_flattened = (right,)
3218 if multi:
3219 return ExpressionClauseList._construct_for_list(
3220 op,
3221 type_,
3222 *(left_flattened + right_flattened),
3223 )
3225 if left._is_collection_aggregate or right._is_collection_aggregate:
3226 negate = None
3228 return BinaryExpression(
3229 left, right, op, type_=type_, negate=negate, modifiers=modifiers
3230 )
3233class ExpressionClauseList(OperatorExpression[_T]):
3234 """Describe a list of clauses, separated by an operator,
3235 in a column expression context.
3237 :class:`.ExpressionClauseList` differs from :class:`.ClauseList` in that
3238 it represents a column-oriented DQL expression only, not an open ended
3239 list of anything comma separated.
3241 .. versionadded:: 2.0
3243 """
3245 __visit_name__ = "expression_clauselist"
3247 _traverse_internals: _TraverseInternalsType = [
3248 ("clauses", InternalTraversal.dp_clauseelement_tuple),
3249 ("operator", InternalTraversal.dp_operator),
3250 ]
3252 clauses: typing_Tuple[ColumnElement[Any], ...]
3254 group: bool
3256 def __init__(
3257 self,
3258 operator: OperatorType,
3259 *clauses: _ColumnExpressionArgument[Any],
3260 type_: Optional[_TypeEngineArgument[_T]] = None,
3261 ):
3262 self.operator = operator
3264 self.clauses = tuple(
3265 coercions.expect(
3266 roles.ExpressionElementRole, clause, apply_propagate_attrs=self
3267 )
3268 for clause in clauses
3269 )
3270 self._is_implicitly_boolean = operators.is_boolean(self.operator)
3271 self.type = type_api.to_instance(type_) # type: ignore[assignment]
3273 @property
3274 def _flattened_operator_clauses(
3275 self,
3276 ) -> typing_Tuple[ColumnElement[Any], ...]:
3277 return self.clauses
3279 def __iter__(self) -> Iterator[ColumnElement[Any]]:
3280 return iter(self.clauses)
3282 def __len__(self) -> int:
3283 return len(self.clauses)
3285 @property
3286 def _select_iterable(self) -> _SelectIterable:
3287 return (self,)
3289 @util.ro_non_memoized_property
3290 def _from_objects(self) -> List[FromClause]:
3291 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
3293 def _append_inplace(self, clause: ColumnElement[Any]) -> None:
3294 self.clauses += (clause,)
3296 @classmethod
3297 def _construct_for_list(
3298 cls,
3299 operator: OperatorType,
3300 type_: TypeEngine[_T],
3301 *clauses: ColumnElement[Any],
3302 group: bool = True,
3303 ) -> ExpressionClauseList[_T]:
3304 self = cls.__new__(cls)
3305 self.group = group
3306 if group:
3307 self.clauses = tuple(
3308 c.self_group(against=operator) for c in clauses
3309 )
3310 else:
3311 self.clauses = clauses
3312 self.operator = operator
3313 self.type = type_
3314 for c in clauses:
3315 if c._propagate_attrs:
3316 self._propagate_attrs = c._propagate_attrs
3317 break
3318 return self
3320 def _negate(self) -> Any:
3321 grouped = self.self_group(against=operators.inv)
3322 assert isinstance(grouped, ColumnElement)
3323 return UnaryExpression(grouped, operator=operators.inv)
3326class BooleanClauseList(ExpressionClauseList[bool]):
3327 __visit_name__ = "expression_clauselist"
3328 inherit_cache = True
3330 def __init__(self, *arg, **kw):
3331 raise NotImplementedError(
3332 "BooleanClauseList has a private constructor"
3333 )
3335 @classmethod
3336 def _process_clauses_for_boolean(
3337 cls,
3338 operator: OperatorType,
3339 continue_on: Any,
3340 skip_on: Any,
3341 clauses: Iterable[ColumnElement[Any]],
3342 ) -> typing_Tuple[int, List[ColumnElement[Any]]]:
3343 has_continue_on = None
3345 convert_clauses = []
3347 against = operators._asbool
3348 lcc = 0
3350 for clause in clauses:
3351 if clause is continue_on:
3352 # instance of continue_on, like and_(x, y, True, z), store it
3353 # if we didn't find one already, we will use it if there
3354 # are no other expressions here.
3355 has_continue_on = clause
3356 elif clause is skip_on:
3357 # instance of skip_on, e.g. and_(x, y, False, z), cancels
3358 # the rest out
3359 convert_clauses = [clause]
3360 lcc = 1
3361 break
3362 else:
3363 if not lcc:
3364 lcc = 1
3365 else:
3366 against = operator
3367 # technically this would be len(convert_clauses) + 1
3368 # however this only needs to indicate "greater than one"
3369 lcc = 2
3370 convert_clauses.append(clause)
3372 if not convert_clauses and has_continue_on is not None:
3373 convert_clauses = [has_continue_on]
3374 lcc = 1
3376 return lcc, [c.self_group(against=against) for c in convert_clauses]
3378 @classmethod
3379 def _construct(
3380 cls,
3381 operator: OperatorType,
3382 continue_on: Any,
3383 skip_on: Any,
3384 initial_clause: Any = _NoArg.NO_ARG,
3385 *clauses: Any,
3386 **kw: Any,
3387 ) -> ColumnElement[Any]:
3388 if initial_clause is _NoArg.NO_ARG:
3389 # no elements period. deprecated use case. return an empty
3390 # ClauseList construct that generates nothing unless it has
3391 # elements added to it.
3392 name = operator.__name__
3394 util.warn_deprecated(
3395 f"Invoking {name}() without arguments is deprecated, and "
3396 f"will be disallowed in a future release. For an empty "
3397 f"""{name}() construct, use '{name}({
3398 'true()' if continue_on is True_._singleton else 'false()'
3399 }, *args)' """
3400 f"""or '{name}({
3401 'True' if continue_on is True_._singleton else 'False'
3402 }, *args)'.""",
3403 version="1.4",
3404 )
3405 return cls._construct_raw(operator)
3407 lcc, convert_clauses = cls._process_clauses_for_boolean(
3408 operator,
3409 continue_on,
3410 skip_on,
3411 [
3412 coercions.expect(roles.WhereHavingRole, clause)
3413 for clause in util.coerce_generator_arg(
3414 (initial_clause,) + clauses
3415 )
3416 ],
3417 )
3419 if lcc > 1:
3420 # multiple elements. Return regular BooleanClauseList
3421 # which will link elements against the operator.
3423 flattened_clauses = itertools.chain.from_iterable(
3424 (
3425 (c for c in to_flat._flattened_operator_clauses)
3426 if getattr(to_flat, "operator", None) is operator
3427 else (to_flat,)
3428 )
3429 for to_flat in convert_clauses
3430 )
3432 return cls._construct_raw(operator, flattened_clauses) # type: ignore[arg-type] # noqa: E501
3433 else:
3434 assert lcc
3435 # just one element. return it as a single boolean element,
3436 # not a list and discard the operator.
3437 return convert_clauses[0]
3439 @classmethod
3440 def _construct_for_whereclause(
3441 cls, clauses: Iterable[ColumnElement[Any]]
3442 ) -> Optional[ColumnElement[bool]]:
3443 operator, continue_on, skip_on = (
3444 operators.and_,
3445 True_._singleton,
3446 False_._singleton,
3447 )
3449 lcc, convert_clauses = cls._process_clauses_for_boolean(
3450 operator,
3451 continue_on,
3452 skip_on,
3453 clauses, # these are assumed to be coerced already
3454 )
3456 if lcc > 1:
3457 # multiple elements. Return regular BooleanClauseList
3458 # which will link elements against the operator.
3459 return cls._construct_raw(operator, convert_clauses)
3460 elif lcc == 1:
3461 # just one element. return it as a single boolean element,
3462 # not a list and discard the operator.
3463 return convert_clauses[0]
3464 else:
3465 return None
3467 @classmethod
3468 def _construct_raw(
3469 cls,
3470 operator: OperatorType,
3471 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
3472 ) -> BooleanClauseList:
3473 self = cls.__new__(cls)
3474 self.clauses = tuple(clauses) if clauses else ()
3475 self.group = True
3476 self.operator = operator
3477 self.type = type_api.BOOLEANTYPE
3478 self._is_implicitly_boolean = True
3479 return self
3481 @classmethod
3482 def and_(
3483 cls,
3484 initial_clause: Union[
3485 Literal[True], _ColumnExpressionArgument[bool], _NoArg
3486 ] = _NoArg.NO_ARG,
3487 *clauses: _ColumnExpressionArgument[bool],
3488 ) -> ColumnElement[bool]:
3489 r"""Produce a conjunction of expressions joined by ``AND``.
3491 See :func:`_sql.and_` for full documentation.
3492 """
3493 return cls._construct(
3494 operators.and_,
3495 True_._singleton,
3496 False_._singleton,
3497 initial_clause,
3498 *clauses,
3499 )
3501 @classmethod
3502 def or_(
3503 cls,
3504 initial_clause: Union[
3505 Literal[False], _ColumnExpressionArgument[bool], _NoArg
3506 ] = _NoArg.NO_ARG,
3507 *clauses: _ColumnExpressionArgument[bool],
3508 ) -> ColumnElement[bool]:
3509 """Produce a conjunction of expressions joined by ``OR``.
3511 See :func:`_sql.or_` for full documentation.
3512 """
3513 return cls._construct(
3514 operators.or_,
3515 False_._singleton,
3516 True_._singleton,
3517 initial_clause,
3518 *clauses,
3519 )
3521 @property
3522 def _select_iterable(self) -> _SelectIterable:
3523 return (self,)
3525 def self_group(
3526 self, against: Optional[OperatorType] = None
3527 ) -> Union[Self, Grouping[bool]]:
3528 if not self.clauses:
3529 return self
3530 else:
3531 return super().self_group(against=against)
3534and_ = BooleanClauseList.and_
3535or_ = BooleanClauseList.or_
3538class Tuple(ClauseList, ColumnElement[TupleAny]):
3539 """Represent a SQL tuple."""
3541 __visit_name__ = "tuple"
3543 _traverse_internals: _TraverseInternalsType = (
3544 ClauseList._traverse_internals + []
3545 )
3547 type: TupleType
3549 @util.preload_module("sqlalchemy.sql.sqltypes")
3550 def __init__(
3551 self,
3552 *clauses: _ColumnExpressionArgument[Any],
3553 types: Optional[Sequence[_TypeEngineArgument[Any]]] = None,
3554 ):
3555 sqltypes = util.preloaded.sql_sqltypes
3557 if types is None:
3558 init_clauses: List[ColumnElement[Any]] = [
3559 coercions.expect(roles.ExpressionElementRole, c)
3560 for c in clauses
3561 ]
3562 else:
3563 if len(types) != len(clauses):
3564 raise exc.ArgumentError(
3565 "Wrong number of elements for %d-tuple: %r "
3566 % (len(types), clauses)
3567 )
3568 init_clauses = [
3569 coercions.expect(
3570 roles.ExpressionElementRole,
3571 c,
3572 type_=typ if not typ._isnull else None,
3573 )
3574 for typ, c in zip(types, clauses)
3575 ]
3577 self.type = sqltypes.TupleType(*[arg.type for arg in init_clauses])
3578 super().__init__(*init_clauses)
3580 @property
3581 def _select_iterable(self) -> _SelectIterable:
3582 return (self,)
3584 def _bind_param(self, operator, obj, type_=None, expanding=False):
3585 if expanding:
3586 return BindParameter(
3587 None,
3588 value=obj,
3589 _compared_to_operator=operator,
3590 unique=True,
3591 expanding=True,
3592 type_=type_,
3593 _compared_to_type=self.type,
3594 )
3595 else:
3596 return Tuple(
3597 *[
3598 BindParameter(
3599 None,
3600 o,
3601 _compared_to_operator=operator,
3602 _compared_to_type=compared_to_type,
3603 unique=True,
3604 type_=type_,
3605 )
3606 for o, compared_to_type in zip(obj, self.type.types)
3607 ]
3608 )
3610 def self_group(self, against: Optional[OperatorType] = None) -> Self:
3611 # Tuple is parenthesized by definition.
3612 return self
3615class Case(ColumnElement[_T]):
3616 """Represent a ``CASE`` expression.
3618 :class:`.Case` is produced using the :func:`.case` factory function,
3619 as in::
3621 from sqlalchemy import case
3623 stmt = select(users_table).where(
3624 case(
3625 (users_table.c.name == "wendy", "W"),
3626 (users_table.c.name == "jack", "J"),
3627 else_="E",
3628 )
3629 )
3631 Details on :class:`.Case` usage is at :func:`.case`.
3633 .. seealso::
3635 :func:`.case`
3637 """
3639 __visit_name__ = "case"
3641 _traverse_internals: _TraverseInternalsType = [
3642 ("value", InternalTraversal.dp_clauseelement),
3643 ("whens", InternalTraversal.dp_clauseelement_tuples),
3644 ("else_", InternalTraversal.dp_clauseelement),
3645 ]
3647 # for case(), the type is derived from the whens. so for the moment
3648 # users would have to cast() the case to get a specific type
3650 whens: List[typing_Tuple[ColumnElement[bool], ColumnElement[_T]]]
3651 else_: Optional[ColumnElement[_T]]
3652 value: Optional[ColumnElement[Any]]
3654 def __init__(
3655 self,
3656 *whens: Union[
3657 typing_Tuple[_ColumnExpressionArgument[bool], Any],
3658 Mapping[Any, Any],
3659 ],
3660 value: Optional[Any] = None,
3661 else_: Optional[Any] = None,
3662 ):
3663 new_whens: Iterable[Any] = coercions._expression_collection_was_a_list(
3664 "whens", "case", whens
3665 )
3666 try:
3667 new_whens = util.dictlike_iteritems(new_whens)
3668 except TypeError:
3669 pass
3671 self.whens = [
3672 (
3673 coercions.expect(
3674 roles.ExpressionElementRole,
3675 c,
3676 apply_propagate_attrs=self,
3677 ).self_group(),
3678 coercions.expect(roles.ExpressionElementRole, r),
3679 )
3680 for (c, r) in new_whens
3681 ]
3683 if value is None:
3684 self.value = None
3685 else:
3686 self.value = coercions.expect(roles.ExpressionElementRole, value)
3688 if else_ is not None:
3689 self.else_ = coercions.expect(roles.ExpressionElementRole, else_)
3690 else:
3691 self.else_ = None
3693 type_ = next(
3694 (
3695 then.type
3696 # Iterate `whens` in reverse to match previous behaviour
3697 # where type of final element took priority
3698 for *_, then in reversed(self.whens)
3699 if not then.type._isnull
3700 ),
3701 self.else_.type if self.else_ is not None else type_api.NULLTYPE,
3702 )
3703 self.type = cast(_T, type_)
3705 @util.ro_non_memoized_property
3706 def _from_objects(self) -> List[FromClause]:
3707 return list(
3708 itertools.chain(*[x._from_objects for x in self.get_children()])
3709 )
3712class Cast(WrapsColumnExpression[_T]):
3713 """Represent a ``CAST`` expression.
3715 :class:`.Cast` is produced using the :func:`.cast` factory function,
3716 as in::
3718 from sqlalchemy import cast, Numeric
3720 stmt = select(cast(product_table.c.unit_price, Numeric(10, 4)))
3722 Details on :class:`.Cast` usage is at :func:`.cast`.
3724 .. seealso::
3726 :ref:`tutorial_casts`
3728 :func:`.cast`
3730 :func:`.try_cast`
3732 :func:`.type_coerce` - an alternative to CAST that coerces the type
3733 on the Python side only, which is often sufficient to generate the
3734 correct SQL and data coercion.
3736 """
3738 __visit_name__ = "cast"
3740 _traverse_internals: _TraverseInternalsType = [
3741 ("clause", InternalTraversal.dp_clauseelement),
3742 ("type", InternalTraversal.dp_type),
3743 ]
3745 clause: ColumnElement[Any]
3746 type: TypeEngine[_T]
3747 typeclause: TypeClause
3749 def __init__(
3750 self,
3751 expression: _ColumnExpressionArgument[Any],
3752 type_: _CovariantTypeEngineArgument[_T],
3753 ):
3754 self.type = type_api.to_instance(type_)
3755 self.clause = coercions.expect(
3756 roles.ExpressionElementRole,
3757 expression,
3758 type_=self.type,
3759 apply_propagate_attrs=self,
3760 )
3761 self.typeclause = TypeClause(self.type)
3763 @util.ro_non_memoized_property
3764 def _from_objects(self) -> List[FromClause]:
3765 return self.clause._from_objects
3767 @property
3768 def wrapped_column_expression(self):
3769 return self.clause
3772class TryCast(Cast[_T]):
3773 """Represent a TRY_CAST expression.
3775 Details on :class:`.TryCast` usage is at :func:`.try_cast`.
3777 .. seealso::
3779 :func:`.try_cast`
3781 :ref:`tutorial_casts`
3782 """
3784 __visit_name__ = "try_cast"
3785 inherit_cache = True
3788class TypeCoerce(WrapsColumnExpression[_T]):
3789 """Represent a Python-side type-coercion wrapper.
3791 :class:`.TypeCoerce` supplies the :func:`_expression.type_coerce`
3792 function; see that function for usage details.
3794 .. seealso::
3796 :func:`_expression.type_coerce`
3798 :func:`.cast`
3800 """
3802 __visit_name__ = "type_coerce"
3804 _traverse_internals: _TraverseInternalsType = [
3805 ("clause", InternalTraversal.dp_clauseelement),
3806 ("type", InternalTraversal.dp_type),
3807 ]
3809 clause: ColumnElement[Any]
3810 type: TypeEngine[_T]
3812 def __init__(
3813 self,
3814 expression: _ColumnExpressionArgument[Any],
3815 type_: _CovariantTypeEngineArgument[_T],
3816 ):
3817 self.type = type_api.to_instance(type_)
3818 self.clause = coercions.expect(
3819 roles.ExpressionElementRole,
3820 expression,
3821 type_=self.type,
3822 apply_propagate_attrs=self,
3823 )
3825 @util.ro_non_memoized_property
3826 def _from_objects(self) -> List[FromClause]:
3827 return self.clause._from_objects
3829 @HasMemoized.memoized_attribute
3830 def typed_expression(self):
3831 if isinstance(self.clause, BindParameter):
3832 bp = self.clause._clone()
3833 bp.type = self.type
3834 return bp
3835 else:
3836 return self.clause
3838 @property
3839 def wrapped_column_expression(self):
3840 return self.clause
3842 def self_group(
3843 self, against: Optional[OperatorType] = None
3844 ) -> TypeCoerce[_T]:
3845 grouped = self.clause.self_group(against=against)
3846 if grouped is not self.clause:
3847 return TypeCoerce(grouped, self.type)
3848 else:
3849 return self
3852class Extract(ColumnElement[int]):
3853 """Represent a SQL EXTRACT clause, ``extract(field FROM expr)``."""
3855 __visit_name__ = "extract"
3857 _traverse_internals: _TraverseInternalsType = [
3858 ("expr", InternalTraversal.dp_clauseelement),
3859 ("field", InternalTraversal.dp_string),
3860 ]
3862 expr: ColumnElement[Any]
3863 field: str
3865 def __init__(self, field: str, expr: _ColumnExpressionArgument[Any]):
3866 self.type = type_api.INTEGERTYPE
3867 self.field = field
3868 self.expr = coercions.expect(roles.ExpressionElementRole, expr)
3870 @util.ro_non_memoized_property
3871 def _from_objects(self) -> List[FromClause]:
3872 return self.expr._from_objects
3875class _label_reference(ColumnElement[_T]):
3876 """Wrap a column expression as it appears in a 'reference' context.
3878 This expression is any that includes an _order_by_label_element,
3879 which is a Label, or a DESC / ASC construct wrapping a Label.
3881 The production of _label_reference() should occur when an expression
3882 is added to this context; this includes the ORDER BY or GROUP BY of a
3883 SELECT statement, as well as a few other places, such as the ORDER BY
3884 within an OVER clause.
3886 """
3888 __visit_name__ = "label_reference"
3890 _traverse_internals: _TraverseInternalsType = [
3891 ("element", InternalTraversal.dp_clauseelement)
3892 ]
3894 element: ColumnElement[_T]
3896 def __init__(self, element: ColumnElement[_T]):
3897 self.element = element
3898 self._propagate_attrs = element._propagate_attrs
3900 @util.ro_non_memoized_property
3901 def _from_objects(self) -> List[FromClause]:
3902 return []
3905class _textual_label_reference(ColumnElement[Any]):
3906 __visit_name__ = "textual_label_reference"
3908 _traverse_internals: _TraverseInternalsType = [
3909 ("element", InternalTraversal.dp_string)
3910 ]
3912 def __init__(self, element: str):
3913 self.element = element
3915 @util.memoized_property
3916 def _text_clause(self) -> TextClause:
3917 return TextClause(self.element)
3920class UnaryExpression(ColumnElement[_T]):
3921 """Define a 'unary' expression.
3923 A unary expression has a single column expression
3924 and an operator. The operator can be placed on the left
3925 (where it is called the 'operator') or right (where it is called the
3926 'modifier') of the column expression.
3928 :class:`.UnaryExpression` is the basis for several unary operators
3929 including those used by :func:`.desc`, :func:`.asc`, :func:`.distinct`,
3930 :func:`.nulls_first` and :func:`.nulls_last`.
3932 """
3934 __visit_name__ = "unary"
3936 _traverse_internals: _TraverseInternalsType = [
3937 ("element", InternalTraversal.dp_clauseelement),
3938 ("operator", InternalTraversal.dp_operator),
3939 ("modifier", InternalTraversal.dp_operator),
3940 ]
3942 element: ColumnElement[Any]
3943 operator: Optional[OperatorType]
3944 modifier: Optional[OperatorType]
3946 def __init__(
3947 self,
3948 element: ColumnElement[Any],
3949 *,
3950 operator: Optional[OperatorType] = None,
3951 modifier: Optional[OperatorType] = None,
3952 type_: Optional[_TypeEngineArgument[_T]] = None,
3953 wraps_column_expression: bool = False, # legacy, not used as of 2.0.42
3954 ):
3955 self.operator = operator
3956 self.modifier = modifier
3957 self._propagate_attrs = element._propagate_attrs
3958 self.element = element.self_group(
3959 against=self.operator or self.modifier
3960 )
3962 # if type is None, we get NULLTYPE, which is our _T. But I don't
3963 # know how to get the overloads to express that correctly
3964 self.type = type_api.to_instance(type_) # type: ignore[assignment]
3966 def _wraps_unnamed_column(self):
3967 ungrouped = self.element._ungroup()
3968 return (
3969 not isinstance(ungrouped, NamedColumn)
3970 or ungrouped._non_anon_label is None
3971 )
3973 @classmethod
3974 def _create_nulls_first(
3975 cls,
3976 column: _ColumnExpressionArgument[_T],
3977 ) -> UnaryExpression[_T]:
3978 return UnaryExpression(
3979 coercions.expect(roles.ByOfRole, column),
3980 modifier=operators.nulls_first_op,
3981 )
3983 @classmethod
3984 def _create_nulls_last(
3985 cls,
3986 column: _ColumnExpressionArgument[_T],
3987 ) -> UnaryExpression[_T]:
3988 return UnaryExpression(
3989 coercions.expect(roles.ByOfRole, column),
3990 modifier=operators.nulls_last_op,
3991 )
3993 @classmethod
3994 def _create_desc(
3995 cls, column: _ColumnExpressionOrStrLabelArgument[_T]
3996 ) -> UnaryExpression[_T]:
3998 return UnaryExpression(
3999 coercions.expect(roles.ByOfRole, column),
4000 modifier=operators.desc_op,
4001 )
4003 @classmethod
4004 def _create_asc(
4005 cls,
4006 column: _ColumnExpressionOrStrLabelArgument[_T],
4007 ) -> UnaryExpression[_T]:
4008 return UnaryExpression(
4009 coercions.expect(roles.ByOfRole, column),
4010 modifier=operators.asc_op,
4011 )
4013 @classmethod
4014 def _create_distinct(
4015 cls,
4016 expr: _ColumnExpressionArgument[_T],
4017 ) -> UnaryExpression[_T]:
4018 col_expr: ColumnElement[_T] = coercions.expect(
4019 roles.ExpressionElementRole, expr
4020 )
4021 return UnaryExpression(
4022 col_expr,
4023 operator=operators.distinct_op,
4024 type_=col_expr.type,
4025 )
4027 @classmethod
4028 def _create_bitwise_not(
4029 cls,
4030 expr: _ColumnExpressionArgument[_T],
4031 ) -> UnaryExpression[_T]:
4032 col_expr: ColumnElement[_T] = coercions.expect(
4033 roles.ExpressionElementRole, expr
4034 )
4035 return UnaryExpression(
4036 col_expr,
4037 operator=operators.bitwise_not_op,
4038 type_=col_expr.type,
4039 )
4041 @property
4042 def _order_by_label_element(self) -> Optional[Label[Any]]:
4043 if operators.is_order_by_modifier(self.modifier):
4044 return self.element._order_by_label_element
4045 else:
4046 return None
4048 @util.ro_non_memoized_property
4049 def _from_objects(self) -> List[FromClause]:
4050 return self.element._from_objects
4052 def _negate(self) -> ColumnElement[Any]:
4053 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
4054 return UnaryExpression(
4055 self.self_group(against=operators.inv),
4056 operator=operators.inv,
4057 type_=type_api.BOOLEANTYPE,
4058 )
4059 else:
4060 return ColumnElement._negate(self)
4062 def self_group(
4063 self, against: Optional[OperatorType] = None
4064 ) -> Union[Self, Grouping[_T]]:
4065 if self.operator and operators.is_precedent(self.operator, against):
4066 return Grouping(self)
4067 else:
4068 return self
4071class CollectionAggregate(UnaryExpression[_T]):
4072 """Forms the basis for right-hand collection operator modifiers
4073 ANY and ALL.
4075 The ANY and ALL keywords are available in different ways on different
4076 backends. On PostgreSQL, they only work for an ARRAY type. On
4077 MySQL, they only work for subqueries.
4079 """
4081 inherit_cache = True
4082 _is_collection_aggregate = True
4084 @classmethod
4085 def _create_any(
4086 cls, expr: _ColumnExpressionArgument[_T]
4087 ) -> CollectionAggregate[bool]:
4088 """create CollectionAggregate for the legacy
4089 ARRAY.Comparator.any() method"""
4090 col_expr: ColumnElement[_T] = coercions.expect(
4091 roles.ExpressionElementRole,
4092 expr,
4093 )
4094 col_expr = col_expr.self_group()
4095 return CollectionAggregate(
4096 col_expr,
4097 operator=operators.any_op,
4098 type_=type_api.BOOLEANTYPE,
4099 )
4101 @classmethod
4102 def _create_all(
4103 cls, expr: _ColumnExpressionArgument[_T]
4104 ) -> CollectionAggregate[bool]:
4105 """create CollectionAggregate for the legacy
4106 ARRAY.Comparator.all() method"""
4107 col_expr: ColumnElement[_T] = coercions.expect(
4108 roles.ExpressionElementRole,
4109 expr,
4110 )
4111 col_expr = col_expr.self_group()
4112 return CollectionAggregate(
4113 col_expr,
4114 operator=operators.all_op,
4115 type_=type_api.BOOLEANTYPE,
4116 )
4118 @util.preload_module("sqlalchemy.sql.sqltypes")
4119 def _bind_param(
4120 self,
4121 operator: operators.OperatorType,
4122 obj: Any,
4123 type_: Optional[TypeEngine[_T]] = None,
4124 expanding: bool = False,
4125 ) -> BindParameter[_T]:
4126 """For new style any_(), all_(), ensure compared literal value
4127 receives appropriate bound parameter type."""
4129 # a CollectionAggregate is specific to ARRAY or int
4130 # only. So for ARRAY case, make sure we use correct element type
4131 sqltypes = util.preloaded.sql_sqltypes
4132 if self.element.type._type_affinity is sqltypes.ARRAY:
4133 compared_to_type = cast(
4134 sqltypes.ARRAY[Any], self.element.type
4135 ).item_type
4136 else:
4137 compared_to_type = self.element.type
4139 return BindParameter(
4140 None,
4141 obj,
4142 _compared_to_operator=operator,
4143 type_=type_,
4144 _compared_to_type=compared_to_type,
4145 unique=True,
4146 expanding=expanding,
4147 )
4149 # operate and reverse_operate are hardwired to
4150 # dispatch onto the type comparator directly, so that we can
4151 # ensure "reversed" behavior.
4152 def operate(
4153 self, op: OperatorType, *other: Any, **kwargs: Any
4154 ) -> ColumnElement[_T]:
4155 if not operators.is_comparison(op):
4156 raise exc.ArgumentError(
4157 "Only comparison operators may be used with ANY/ALL"
4158 )
4159 kwargs["reverse"] = True
4160 return self.comparator.operate(operators.mirror(op), *other, **kwargs)
4162 def reverse_operate(
4163 self, op: OperatorType, other: Any, **kwargs: Any
4164 ) -> ColumnElement[_T]:
4165 # comparison operators should never call reverse_operate
4166 assert not operators.is_comparison(op)
4167 raise exc.ArgumentError(
4168 "Only comparison operators may be used with ANY/ALL"
4169 )
4172class AsBoolean(WrapsColumnExpression[bool], UnaryExpression[bool]):
4173 inherit_cache = True
4175 def __init__(self, element, operator, negate):
4176 self.element = element
4177 self.type = type_api.BOOLEANTYPE
4178 self.operator = operator
4179 self.negate = negate
4180 self.modifier = None
4181 self._is_implicitly_boolean = element._is_implicitly_boolean
4183 @property
4184 def wrapped_column_expression(self):
4185 return self.element
4187 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4188 return self
4190 def _negate(self):
4191 if isinstance(self.element, (True_, False_)):
4192 return self.element._negate()
4193 else:
4194 return AsBoolean(self.element, self.negate, self.operator)
4197class BinaryExpression(OperatorExpression[_T]):
4198 """Represent an expression that is ``LEFT <operator> RIGHT``.
4200 A :class:`.BinaryExpression` is generated automatically
4201 whenever two column expressions are used in a Python binary expression:
4203 .. sourcecode:: pycon+sql
4205 >>> from sqlalchemy.sql import column
4206 >>> column("a") + column("b")
4207 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
4208 >>> print(column("a") + column("b"))
4209 {printsql}a + b
4211 """
4213 __visit_name__ = "binary"
4215 _traverse_internals: _TraverseInternalsType = [
4216 ("left", InternalTraversal.dp_clauseelement),
4217 ("right", InternalTraversal.dp_clauseelement),
4218 ("operator", InternalTraversal.dp_operator),
4219 ("negate", InternalTraversal.dp_operator),
4220 ("modifiers", InternalTraversal.dp_plain_dict),
4221 (
4222 "type",
4223 InternalTraversal.dp_type,
4224 ),
4225 ]
4227 _cache_key_traversal = [
4228 ("left", InternalTraversal.dp_clauseelement),
4229 ("right", InternalTraversal.dp_clauseelement),
4230 ("operator", InternalTraversal.dp_operator),
4231 ("modifiers", InternalTraversal.dp_plain_dict),
4232 # "type" affects JSON CAST operators, so while redundant in most cases,
4233 # is needed for that one
4234 (
4235 "type",
4236 InternalTraversal.dp_type,
4237 ),
4238 ]
4240 _is_implicitly_boolean = True
4241 """Indicates that any database will know this is a boolean expression
4242 even if the database does not have an explicit boolean datatype.
4244 """
4246 left: ColumnElement[Any]
4247 right: ColumnElement[Any]
4248 modifiers: Mapping[str, Any]
4250 def __init__(
4251 self,
4252 left: ColumnElement[Any],
4253 right: ColumnElement[Any],
4254 operator: OperatorType,
4255 type_: Optional[_TypeEngineArgument[_T]] = None,
4256 negate: Optional[OperatorType] = None,
4257 modifiers: Optional[Mapping[str, Any]] = None,
4258 ):
4259 # allow compatibility with libraries that
4260 # refer to BinaryExpression directly and pass strings
4261 if isinstance(operator, str):
4262 operator = operators.custom_op(operator)
4263 self._orig = (left.__hash__(), right.__hash__())
4264 self._propagate_attrs = left._propagate_attrs or right._propagate_attrs
4265 self.left = left.self_group(against=operator)
4266 self.right = right.self_group(against=operator)
4267 self.operator = operator
4269 # if type is None, we get NULLTYPE, which is our _T. But I don't
4270 # know how to get the overloads to express that correctly
4271 self.type = type_api.to_instance(type_) # type: ignore[assignment]
4273 self.negate = negate
4274 self._is_implicitly_boolean = operators.is_boolean(operator)
4276 if modifiers is None:
4277 self.modifiers = {}
4278 else:
4279 self.modifiers = modifiers
4281 @property
4282 def _flattened_operator_clauses(
4283 self,
4284 ) -> typing_Tuple[ColumnElement[Any], ...]:
4285 return (self.left, self.right)
4287 def __bool__(self):
4288 """Implement Python-side "bool" for BinaryExpression as a
4289 simple "identity" check for the left and right attributes,
4290 if the operator is "eq" or "ne". Otherwise the expression
4291 continues to not support "bool" like all other column expressions.
4293 The rationale here is so that ColumnElement objects can be hashable.
4294 What? Well, suppose you do this::
4296 c1, c2 = column("x"), column("y")
4297 s1 = set([c1, c2])
4299 We do that **a lot**, columns inside of sets is an extremely basic
4300 thing all over the ORM for example.
4302 So what happens if we do this? ::
4304 c1 in s1
4306 Hashing means it will normally use ``__hash__()`` of the object,
4307 but in case of hash collision, it's going to also do ``c1 == c1``
4308 and/or ``c1 == c2`` inside. Those operations need to return a
4309 True/False value. But because we override ``==`` and ``!=``, they're
4310 going to get a BinaryExpression. Hence we implement ``__bool__`` here
4311 so that these comparisons behave in this particular context mostly
4312 like regular object comparisons. Thankfully Python is OK with
4313 that! Otherwise we'd have to use special set classes for columns
4314 (which we used to do, decades ago).
4316 """
4317 if self.operator in (operators.eq, operators.ne):
4318 # this is using the eq/ne operator given int hash values,
4319 # rather than Operator, so that "bool" can be based on
4320 # identity
4321 return self.operator(*self._orig) # type: ignore[call-overload]
4322 else:
4323 raise TypeError("Boolean value of this clause is not defined")
4325 if typing.TYPE_CHECKING:
4327 def __invert__(
4328 self: BinaryExpression[_T],
4329 ) -> BinaryExpression[_T]: ...
4331 @util.ro_non_memoized_property
4332 def _from_objects(self) -> List[FromClause]:
4333 return self.left._from_objects + self.right._from_objects
4335 def _negate(self):
4336 if self.negate is not None:
4337 return BinaryExpression(
4338 self.left,
4339 self.right._negate_in_binary(self.negate, self.operator),
4340 self.negate,
4341 negate=self.operator,
4342 type_=self.type,
4343 modifiers=self.modifiers,
4344 )
4345 else:
4346 return self.self_group()._negate()
4349class Slice(ColumnElement[Any]):
4350 """Represent SQL for a Python array-slice object.
4352 This is not a specific SQL construct at this level, but
4353 may be interpreted by specific dialects, e.g. PostgreSQL.
4355 """
4357 __visit_name__ = "slice"
4359 _traverse_internals: _TraverseInternalsType = [
4360 ("start", InternalTraversal.dp_clauseelement),
4361 ("stop", InternalTraversal.dp_clauseelement),
4362 ("step", InternalTraversal.dp_clauseelement),
4363 ]
4365 def __init__(self, start, stop, step, _name=None):
4366 self.start = coercions.expect(
4367 roles.ExpressionElementRole,
4368 start,
4369 name=_name,
4370 type_=type_api.INTEGERTYPE,
4371 )
4372 self.stop = coercions.expect(
4373 roles.ExpressionElementRole,
4374 stop,
4375 name=_name,
4376 type_=type_api.INTEGERTYPE,
4377 )
4378 self.step = coercions.expect(
4379 roles.ExpressionElementRole,
4380 step,
4381 name=_name,
4382 type_=type_api.INTEGERTYPE,
4383 )
4384 self.type = type_api.NULLTYPE
4386 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4387 assert against is operator.getitem
4388 return self
4391class IndexExpression(BinaryExpression[Any]):
4392 """Represent the class of expressions that are like an "index"
4393 operation."""
4395 inherit_cache = True
4398class GroupedElement(DQLDMLClauseElement):
4399 """Represent any parenthesized expression"""
4401 __visit_name__ = "grouping"
4403 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4404 return self
4406 def _ungroup(self) -> ClauseElement:
4407 raise NotImplementedError()
4410class Grouping(GroupedElement, ColumnElement[_T]):
4411 """Represent a grouping within a column expression"""
4413 _traverse_internals: _TraverseInternalsType = [
4414 ("element", InternalTraversal.dp_clauseelement),
4415 ("type", InternalTraversal.dp_type),
4416 ]
4418 _cache_key_traversal = [
4419 ("element", InternalTraversal.dp_clauseelement),
4420 ]
4422 element: Union[
4423 AbstractTextClause,
4424 ClauseList,
4425 ColumnElement[_T],
4426 CompilerColumnElement,
4427 ]
4429 def __init__(
4430 self,
4431 element: Union[
4432 AbstractTextClause,
4433 ClauseList,
4434 ColumnElement[_T],
4435 CompilerColumnElement,
4436 ],
4437 ):
4438 self.element = element
4440 # nulltype assignment issue
4441 self.type = getattr(element, "type", type_api.NULLTYPE) # type: ignore[arg-type] # noqa: E501
4442 self._propagate_attrs = element._propagate_attrs
4444 def _with_binary_element_type(self, type_):
4445 return self.__class__(self.element._with_binary_element_type(type_))
4447 def _ungroup(self) -> ColumnElement[_T]:
4448 assert isinstance(self.element, ColumnElement)
4449 return self.element._ungroup()
4451 @util.memoized_property
4452 def _is_implicitly_boolean(self):
4453 return self.element._is_implicitly_boolean
4455 @util.non_memoized_property
4456 def _tq_label(self) -> Optional[str]:
4457 return (
4458 getattr(self.element, "_tq_label", None) or self._anon_name_label
4459 )
4461 @util.non_memoized_property
4462 def _proxies(self) -> List[ColumnElement[Any]]:
4463 if isinstance(self.element, ColumnElement):
4464 return [self.element]
4465 else:
4466 return []
4468 @util.ro_non_memoized_property
4469 def _from_objects(self) -> List[FromClause]:
4470 return self.element._from_objects
4472 def __getattr__(self, attr):
4473 return getattr(self.element, attr)
4475 def __getstate__(self):
4476 return {"element": self.element, "type": self.type}
4478 def __setstate__(self, state):
4479 self.element = state["element"]
4480 self.type = state["type"]
4482 if TYPE_CHECKING:
4484 def self_group(
4485 self, against: Optional[OperatorType] = None
4486 ) -> Self: ...
4489class _OverrideBinds(ColumnElement[_T]):
4490 """used by cache_key->_apply_params_to_element to allow compilation /
4491 execution of a SQL element that's been cached, using an alternate set of
4492 bound parameter values.
4494 This is used by the ORM to swap new parameter values into expressions
4495 that are embedded into loader options like with_expression(),
4496 selectinload(). Previously, this task was accomplished using the
4497 .params() method which would perform a deep-copy instead. This deep
4498 copy proved to be too expensive for more complex expressions.
4500 The construct is a stand-in for the element it wraps; it renders, groups
4501 and generates a cache key the same way as that element, differing only in
4502 the bound parameter values it delivers.
4504 See #11085
4506 """
4508 __visit_name__ = "override_binds"
4510 _traverse_internals: _TraverseInternalsType = [
4511 ("element", InternalTraversal.dp_clauseelement),
4512 ("type", InternalTraversal.dp_type),
4513 ]
4515 element: ColumnElement[_T]
4517 def __init__(
4518 self,
4519 element: ColumnElement[_T],
4520 bindparams: Sequence[BindParameter[Any]],
4521 replaces_params: Sequence[BindParameter[Any]],
4522 ):
4523 self.element = element
4525 # take on the type of the element we wrap; otherwise the construct
4526 # reports NULLTYPE and the result set for a column that's compiled
4527 # from here has no result processor
4528 self.type = element.type
4529 self.translate = {
4530 k.key: v.value for k, v in zip(replaces_params, bindparams)
4531 }
4533 def _gen_cache_key(
4534 self, anon_map: anon_map, bindparams: List[BindParameter[Any]]
4535 ) -> Optional[typing_Tuple[Any, ...]]:
4536 """generate a cache key for the given element, substituting its bind
4537 values for the translation values present."""
4539 translate = self.translate
4541 ck = self.element._gen_cache_key(anon_map, bindparams)
4543 # the anon_map check stops self.element's traversal at objects
4544 # already visited, so their bound parameters aren't collected again
4545 # here. this happens when an element visited ahead of this one,
4546 # such as the path of a loader option, refers to the same aliased()
4547 # subquery as the criteria wrapped here. those parameters are
4548 # already in bindparams with the original values, so the
4549 # translation is applied to the whole list. see #13766
4550 bindparams[:] = [
4551 (
4552 bp._with_value(
4553 translate[bp.key], maintain_key=True, required=False
4554 )
4555 if bp.key in translate
4556 else bp
4557 )
4558 for bp in bindparams
4559 ]
4561 # ck derives from _gen_cache_key, a compiled function in
4562 # _cache_key_cy that mypy sees as untyped
4563 return ck # type: ignore[no-any-return]
4565 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4566 # group the element we wrap the same way it would group on its own.
4567 # see #13753
4568 grouped = self.element.self_group(against=against)
4569 if grouped is self.element:
4570 return self
4572 new = self._clone()
4573 new.element = grouped
4574 return new
4576 def _negate(self) -> ColumnElement[Any]:
4577 # negate the element we wrap the same way it would negate on its
4578 # own, e.g. NOT (a OR b) rather than NOT a OR b. see #13753
4579 new = self._clone()
4580 new.element = self.element._negate()
4581 new.type = new.element.type
4582 return new
4584 @util.ro_non_memoized_property
4585 def _from_objects(self) -> List[FromClause]:
4586 return self.element._from_objects
4588 @util.non_memoized_property
4589 def _is_implicitly_boolean(self) -> bool:
4590 return self.element._is_implicitly_boolean
4592 @util.non_memoized_property
4593 def _proxies(self) -> List[ColumnElement[Any]]:
4594 return [self.element]
4596 @util.non_memoized_property
4597 def _tq_label(self) -> Optional[str]:
4598 return (
4599 getattr(self.element, "_tq_label", None) or self._anon_name_label
4600 )
4603_FrameIntTuple = tuple[int | None, int | None]
4606class Over(ColumnElement[_T]):
4607 """Represent an OVER clause.
4609 This is a special operator against a so-called
4610 "window" function, as well as any aggregate function,
4611 which produces results relative to the result set
4612 itself. Most modern SQL backends now support window functions.
4614 """
4616 __visit_name__ = "over"
4618 _traverse_internals: _TraverseInternalsType = [
4619 ("element", InternalTraversal.dp_clauseelement),
4620 ("order_by", InternalTraversal.dp_clauseelement),
4621 ("partition_by", InternalTraversal.dp_clauseelement),
4622 ("range_", InternalTraversal.dp_clauseelement),
4623 ("rows", InternalTraversal.dp_clauseelement),
4624 ("groups", InternalTraversal.dp_clauseelement),
4625 ("exclude", InternalTraversal.dp_string),
4626 ]
4628 order_by: Optional[ClauseList] = None
4629 partition_by: Optional[ClauseList] = None
4631 element: ColumnElement[_T]
4632 """The underlying expression object to which this :class:`.Over`
4633 object refers."""
4635 range_: FrameClause | None
4636 rows: FrameClause | None
4637 groups: FrameClause | None
4638 exclude: str | None
4640 def __init__(
4641 self,
4642 element: ColumnElement[_T],
4643 partition_by: Optional[_ByArgument] = None,
4644 order_by: Optional[_ByArgument] = None,
4645 range_: _FrameIntTuple | FrameClause | None = None,
4646 rows: _FrameIntTuple | FrameClause | None = None,
4647 groups: _FrameIntTuple | FrameClause | None = None,
4648 exclude: str | None = None,
4649 ):
4650 self.element = element
4651 if order_by is not None:
4652 self.order_by = ClauseList(
4653 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4654 )
4655 if partition_by is not None:
4656 self.partition_by = ClauseList(
4657 *util.to_list(partition_by),
4658 _literal_as_text_role=roles.ByOfRole,
4659 )
4661 if sum(item is not None for item in (range_, rows, groups)) > 1:
4662 raise exc.ArgumentError(
4663 "only one of 'rows', 'range_', or 'groups' may be provided"
4664 )
4665 else:
4666 self.range_ = FrameClause._parse(range_, coerce_int=False)
4667 self.rows = FrameClause._parse(rows, coerce_int=True)
4668 self.groups = FrameClause._parse(groups, coerce_int=True)
4670 self.exclude = exclude
4672 if exclude is not None and (
4673 range_ is None and rows is None and groups is None
4674 ):
4675 raise exc.ArgumentError(
4676 "'exclude' requires that one of 'rows', "
4677 "'range_', or 'groups' is also specified"
4678 )
4680 if not TYPE_CHECKING:
4682 @util.memoized_property
4683 def type(self) -> TypeEngine[_T]: # noqa: A001
4684 return self.element.type
4686 @util.ro_non_memoized_property
4687 def _from_objects(self) -> List[FromClause]:
4688 return list(
4689 itertools.chain(
4690 *[
4691 c._from_objects
4692 for c in (self.element, self.partition_by, self.order_by)
4693 if c is not None
4694 ]
4695 )
4696 )
4699class FrameClauseType(Enum):
4700 """Frame clause type enum for FrameClause lower_type and upper_type.
4702 .. versionadded:: 2.1
4704 """
4706 UNBOUNDED = 0
4707 """Produces an "UNBOUNDED PRECEDING" or "UNBOUNDED FOLLOWING" frame
4708 clause depending on the position.
4709 Requires a ``None`` value for the corresponding bound value.
4710 """
4711 CURRENT = 1
4712 """Produces a "CURRENT ROW" frame clause.
4713 Requires a ``None`` value for the corresponding bound value.
4714 """
4715 PRECEDING = 2
4716 """Produces a "PRECEDING" frame clause."""
4717 FOLLOWING = 3
4718 """Produces a "FOLLOWING" frame clause."""
4721_require_none = (
4722 FrameClauseType.CURRENT,
4723 FrameClauseType.UNBOUNDED,
4724)
4727class FrameClause(ClauseElement):
4728 """Indicate the 'rows' 'range' or 'group' field of a window function,
4729 e.g. using :class:`.Over`.
4731 .. versionadded:: 2.1
4733 """
4735 __visit_name__ = "frame_clause"
4737 _traverse_internals: _TraverseInternalsType = [
4738 ("lower_bind", InternalTraversal.dp_clauseelement),
4739 ("upper_bind", InternalTraversal.dp_clauseelement),
4740 ("lower_type", InternalTraversal.dp_plain_obj),
4741 ("upper_type", InternalTraversal.dp_plain_obj),
4742 ]
4744 def __init__(
4745 self,
4746 start: Any,
4747 end: Any,
4748 start_frame_type: FrameClauseType,
4749 end_frame_type: FrameClauseType,
4750 _validate: bool = True,
4751 ) -> None:
4752 """Creates a new FrameClause specifying the bounds of a window frame.
4754 :param start: The start value.
4755 :param end: The end value.
4756 :param start_frame_type: The :class:`FrameClauseType` for the
4757 start value.
4758 :param end_frame_type: The :class:`FrameClauseType` for the end value.
4759 """
4760 self.lower_bind = self._as_literal(start)
4761 self.upper_bind = self._as_literal(end)
4762 self.lower_type = FrameClauseType(start_frame_type)
4763 self.upper_type = FrameClauseType(end_frame_type)
4764 if _validate:
4765 if (
4766 self.lower_type in _require_none
4767 and self.lower_bind is not None
4768 ):
4769 raise exc.ArgumentError(
4770 "Cannot specify a value for start with frame type "
4771 f"{self.lower_type.name}"
4772 )
4773 if (
4774 self.upper_type in _require_none
4775 and self.upper_bind is not None
4776 ):
4777 raise exc.ArgumentError(
4778 "Cannot specify a value for end with frame type "
4779 f"{self.upper_type.name}"
4780 )
4782 @classmethod
4783 def _as_literal(cls, value: Any) -> BindParameter[Any] | None:
4784 if value is None:
4785 return None
4786 elif isinstance(value, int):
4787 return literal(value, type_api.INTEGERTYPE)
4788 elif isinstance(value, BindParameter):
4789 return value
4790 else:
4791 return literal(value) # let the default type resolution occur
4793 @classmethod
4794 def _handle_int(
4795 cls, value: Any | None, coerce_int: bool
4796 ) -> tuple[int | None, FrameClauseType]:
4797 if value is None:
4798 return None, FrameClauseType.UNBOUNDED
4800 if coerce_int:
4801 try:
4802 integer = int(value)
4803 except ValueError as err:
4804 raise exc.ArgumentError(
4805 "Integer or None expected for values in rows/groups frame"
4806 ) from err
4807 elif not isinstance(value, int):
4808 raise exc.ArgumentError(
4809 "When using a tuple to specify a range only integer or none "
4810 "values are allowed in the range frame. To specify a "
4811 "different type use the FrameClause directly."
4812 )
4813 else:
4814 integer = value
4815 if integer == 0:
4816 return None, FrameClauseType.CURRENT
4817 elif integer < 0:
4818 return abs(integer), FrameClauseType.PRECEDING
4819 else:
4820 return integer, FrameClauseType.FOLLOWING
4822 @classmethod
4823 def _parse(
4824 cls,
4825 range_: _FrameIntTuple | FrameClause | None,
4826 coerce_int: bool,
4827 ) -> FrameClause | None:
4828 if range_ is None or isinstance(range_, FrameClause):
4829 return range_
4831 try:
4832 r0, r1 = range_
4833 except (ValueError, TypeError) as ve:
4834 raise exc.ArgumentError(
4835 "2-tuple expected for range/rows/groups"
4836 ) from ve
4838 l_b, l_t = cls._handle_int(r0, coerce_int)
4839 u_b, u_t = cls._handle_int(r1, coerce_int)
4841 return FrameClause(
4842 start=l_b,
4843 end=u_b,
4844 start_frame_type=l_t,
4845 end_frame_type=u_t,
4846 _validate=False,
4847 )
4850class AggregateOrderBy(WrapsColumnExpression[_T]):
4851 """Represent an aggregate ORDER BY expression.
4853 This is a special operator against aggregate functions such as
4854 ``array_agg()``, ``json_arrayagg()`` ``string_agg()``, etc. that provides
4855 for an ORDER BY expression, using a syntax that's compatible with
4856 the backend.
4858 :class:`.AggregateOrderBy` is a generalized version of the
4859 :class:`.WithinGroup` construct, the latter of which always provides a
4860 "WITHIN GROUP (ORDER BY ...)" expression. :class:`.AggregateOrderBy` will
4861 also compile to "WITHIN GROUP (ORDER BY ...)" on backends such as Oracle
4862 and SQL Server that don't have another style of aggregate function
4863 ordering.
4865 .. versionadded:: 2.1
4868 """
4870 __visit_name__ = "aggregateorderby"
4872 _traverse_internals: _TraverseInternalsType = [
4873 ("element", InternalTraversal.dp_clauseelement),
4874 ("order_by", InternalTraversal.dp_clauseelement),
4875 ]
4877 order_by: ClauseList
4879 def __init__(
4880 self,
4881 element: Union[FunctionElement[_T], FunctionFilter[_T]],
4882 *order_by: _ColumnExpressionArgument[Any],
4883 ):
4884 self.element = element
4885 if not order_by:
4886 raise TypeError("at least one ORDER BY element is required")
4887 self.order_by = ClauseList(
4888 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4889 )
4891 if not TYPE_CHECKING:
4893 @util.memoized_property
4894 def type(self) -> TypeEngine[_T]: # noqa: A001
4895 return self.element.type
4897 @property
4898 def wrapped_column_expression(self) -> ColumnElement[_T]:
4899 return self.element
4901 def __reduce__(self):
4902 return self.__class__, (self.element,) + (
4903 tuple(self.order_by) if self.order_by is not None else ()
4904 )
4906 def over(
4907 self,
4908 *,
4909 partition_by: _ByArgument | None = None,
4910 order_by: _ByArgument | None = None,
4911 rows: _FrameIntTuple | FrameClause | None = None,
4912 range_: _FrameIntTuple | FrameClause | None = None,
4913 groups: _FrameIntTuple | FrameClause | None = None,
4914 exclude: str | None = None,
4915 ) -> Over[_T]:
4916 """Produce an OVER clause against this :class:`.WithinGroup`
4917 construct.
4919 This function has the same signature as that of
4920 :meth:`.FunctionElement.over`.
4922 """
4923 return Over(
4924 self,
4925 partition_by=partition_by,
4926 order_by=order_by,
4927 range_=range_,
4928 rows=rows,
4929 groups=groups,
4930 exclude=exclude,
4931 )
4933 @overload
4934 def filter(self) -> Self: ...
4936 @overload
4937 def filter(
4938 self,
4939 __criterion0: _ColumnExpressionArgument[bool],
4940 *criterion: _ColumnExpressionArgument[bool],
4941 ) -> FunctionFilter[_T]: ...
4943 def filter(
4944 self, *criterion: _ColumnExpressionArgument[bool]
4945 ) -> Union[Self, FunctionFilter[_T]]:
4946 """Produce a FILTER clause against this function."""
4947 if not criterion:
4948 return self
4949 return FunctionFilter(self, *criterion)
4951 @util.ro_non_memoized_property
4952 def _from_objects(self) -> List[FromClause]:
4953 return list(
4954 itertools.chain(
4955 *[
4956 c._from_objects
4957 for c in (self.element, self.order_by)
4958 if c is not None
4959 ]
4960 )
4961 )
4964class WithinGroup(AggregateOrderBy[_T]):
4965 """Represent a WITHIN GROUP (ORDER BY) clause.
4967 This is a special operator against so-called
4968 "ordered set aggregate" and "hypothetical
4969 set aggregate" functions, including ``percentile_cont()``,
4970 ``rank()``, ``dense_rank()``, etc.
4972 It's supported only by certain database backends, such as PostgreSQL,
4973 Oracle Database and MS SQL Server.
4975 The :class:`.WithinGroup` construct extracts its type from the
4976 method :meth:`.FunctionElement.within_group_type`. If this returns
4977 ``None``, the function's ``.type`` is used.
4979 """
4981 __visit_name__ = "withingroup"
4982 inherit_cache = True
4984 if not TYPE_CHECKING:
4986 @util.memoized_property
4987 def type(self) -> TypeEngine[_T]: # noqa: A001
4988 wgt = self.element.within_group_type(self)
4989 if wgt is not None:
4990 return wgt
4991 else:
4992 return self.element.type
4995class FunctionFilter(Generative, ColumnElement[_T]):
4996 """Represent a function FILTER clause.
4998 This is a special operator against aggregate and window functions,
4999 which controls which rows are passed to it.
5000 It's supported only by certain database backends.
5002 Invocation of :class:`.FunctionFilter` is via
5003 :meth:`.FunctionElement.filter`::
5005 func.count(1).filter(True)
5007 .. seealso::
5009 :meth:`.FunctionElement.filter`
5011 """
5013 __visit_name__ = "funcfilter"
5015 _traverse_internals: _TraverseInternalsType = [
5016 ("func", InternalTraversal.dp_clauseelement),
5017 ("criterion", InternalTraversal.dp_clauseelement),
5018 ]
5020 criterion: Optional[ColumnElement[bool]] = None
5022 def __init__(
5023 self,
5024 func: Union[FunctionElement[_T], AggregateOrderBy[_T]],
5025 *criterion: _ColumnExpressionArgument[bool],
5026 ):
5027 self.func = func
5028 self.filter.non_generative(self, *criterion) # type: ignore[attr-defined] # noqa: E501
5030 @_generative
5031 def filter(self, *criterion: _ColumnExpressionArgument[bool]) -> Self:
5032 """Produce an additional FILTER against the function.
5034 This method adds additional criteria to the initial criteria
5035 set up by :meth:`.FunctionElement.filter`.
5037 Multiple criteria are joined together at SQL render time
5038 via ``AND``.
5041 """
5043 for crit in list(criterion):
5044 crit = coercions.expect(roles.WhereHavingRole, crit)
5046 if self.criterion is not None:
5047 self.criterion = self.criterion & crit
5048 else:
5049 self.criterion = crit
5051 return self
5053 def over(
5054 self,
5055 partition_by: _ByArgument | None = None,
5056 order_by: _ByArgument | None = None,
5057 range_: _FrameIntTuple | FrameClause | None = None,
5058 rows: _FrameIntTuple | FrameClause | None = None,
5059 groups: _FrameIntTuple | FrameClause | None = None,
5060 exclude: str | None = None,
5061 ) -> Over[_T]:
5062 """Produce an OVER clause against this filtered function.
5064 Used against aggregate or so-called "window" functions,
5065 for database backends that support window functions.
5067 The expression::
5069 func.rank().filter(MyClass.y > 5).over(order_by="x")
5071 is shorthand for::
5073 from sqlalchemy import over, funcfilter
5075 over(funcfilter(func.rank(), MyClass.y > 5), order_by="x")
5077 See :func:`_expression.over` for a full description.
5079 """
5080 return Over(
5081 self,
5082 partition_by=partition_by,
5083 order_by=order_by,
5084 range_=range_,
5085 rows=rows,
5086 groups=groups,
5087 exclude=exclude,
5088 )
5090 def within_group(
5091 self, *order_by: _ColumnExpressionArgument[Any]
5092 ) -> WithinGroup[_T]:
5093 """Produce a WITHIN GROUP (ORDER BY expr) clause against
5094 this function.
5095 """
5096 return WithinGroup(self, *order_by)
5098 def within_group_type(
5099 self, within_group: WithinGroup[_T]
5100 ) -> Optional[TypeEngine[_T]]:
5101 return None
5103 def self_group(
5104 self, against: Optional[OperatorType] = None
5105 ) -> Union[Self, Grouping[_T]]:
5106 if operators.is_precedent(operators.filter_op, against):
5107 return Grouping(self)
5108 else:
5109 return self
5111 if not TYPE_CHECKING:
5113 @util.memoized_property
5114 def type(self) -> TypeEngine[_T]: # noqa: A001
5115 return self.func.type
5117 @util.ro_non_memoized_property
5118 def _from_objects(self) -> List[FromClause]:
5119 return list(
5120 itertools.chain(
5121 *[
5122 c._from_objects
5123 for c in (self.func, self.criterion)
5124 if c is not None
5125 ]
5126 )
5127 )
5130class NamedColumn(KeyedColumnElement[_T]):
5131 is_literal = False
5132 table: Optional[FromClause] = None
5133 name: str
5134 key: str
5136 def _compare_name_for_result(self, other):
5137 return (hasattr(other, "name") and self.name == other.name) or (
5138 hasattr(other, "_label") and self._label == other._label
5139 )
5141 @util.ro_memoized_property
5142 def description(self) -> str:
5143 return self.name
5145 @HasMemoized.memoized_attribute
5146 def _tq_key_label(self) -> Optional[str]:
5147 """table qualified label based on column key.
5149 for table-bound columns this is <tablename>_<column key/proxy key>;
5151 all other expressions it resolves to key/proxy key.
5153 """
5154 proxy_key = self._proxy_key
5155 if proxy_key and proxy_key != self.name:
5156 return self._gen_tq_label(proxy_key)
5157 else:
5158 return self._tq_label
5160 @HasMemoized.memoized_attribute
5161 def _tq_label(self) -> Optional[str]:
5162 """table qualified label based on column name.
5164 for table-bound columns this is <tablename>_<columnname>; all other
5165 expressions it resolves to .name.
5167 """
5168 return self._gen_tq_label(self.name)
5170 @HasMemoized.memoized_attribute
5171 def _render_label_in_columns_clause(self):
5172 return True
5174 @HasMemoized.memoized_attribute
5175 def _non_anon_label(self):
5176 return self.name
5178 def _gen_tq_label(
5179 self, name: str, dedupe_on_key: bool = True
5180 ) -> Optional[str]:
5181 return name
5183 def _bind_param(
5184 self,
5185 operator: OperatorType,
5186 obj: Any,
5187 type_: Optional[TypeEngine[_T]] = None,
5188 expanding: bool = False,
5189 ) -> BindParameter[_T]:
5190 return BindParameter(
5191 self.key,
5192 obj,
5193 _compared_to_operator=operator,
5194 _compared_to_type=self.type,
5195 type_=type_,
5196 unique=True,
5197 expanding=expanding,
5198 )
5200 def _make_proxy(
5201 self,
5202 selectable: FromClause,
5203 *,
5204 primary_key: ColumnSet,
5205 foreign_keys: Set[KeyedColumnElement[Any]],
5206 name: Optional[str] = None,
5207 key: Optional[str] = None,
5208 name_is_truncatable: bool = False,
5209 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5210 disallow_is_literal: bool = False,
5211 **kw: Any,
5212 ) -> typing_Tuple[str, ColumnClause[_T]]:
5213 c = ColumnClause(
5214 (
5215 coercions.expect(roles.TruncatedLabelRole, name or self.name)
5216 if name_is_truncatable
5217 else (name or self.name)
5218 ),
5219 type_=self.type,
5220 _selectable=selectable,
5221 is_literal=False,
5222 )
5224 c._propagate_attrs = selectable._propagate_attrs
5225 if name is None:
5226 c.key = self.key
5227 if compound_select_cols:
5228 c._proxies = list(compound_select_cols)
5229 else:
5230 c._proxies = [self]
5232 if selectable._is_clone_of is not None:
5233 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5234 return c.key, c
5237_PS = ParamSpec("_PS")
5240class Label(roles.LabeledColumnExprRole[_T], NamedColumn[_T]):
5241 """Represents a column label (AS).
5243 Represent a label, as typically applied to any column-level
5244 element using the ``AS`` sql keyword.
5246 """
5248 __visit_name__ = "label"
5250 _traverse_internals: _TraverseInternalsType = [
5251 ("name", InternalTraversal.dp_anon_name),
5252 ("type", InternalTraversal.dp_type),
5253 ("_element", InternalTraversal.dp_clauseelement),
5254 ]
5256 _cache_key_traversal = [
5257 ("name", InternalTraversal.dp_anon_name),
5258 ("_element", InternalTraversal.dp_clauseelement),
5259 ]
5261 _element: ColumnElement[_T]
5262 name: str
5264 def __init__(
5265 self,
5266 name: Optional[str],
5267 element: _ColumnExpressionArgument[_T],
5268 type_: Optional[_CovariantTypeEngineArgument[_T]] = None,
5269 ):
5270 orig_element = element
5271 element = coercions.expect(
5272 roles.ExpressionElementRole,
5273 element,
5274 apply_propagate_attrs=self,
5275 )
5276 while isinstance(element, Label):
5277 # TODO: this is only covered in test_text.py, but nothing
5278 # fails if it's removed. determine rationale
5279 element = element.element
5281 if name:
5282 self.name = name
5283 else:
5284 self.name = _anonymous_label.safe_construct(
5285 id(self), getattr(element, "name", "anon")
5286 )
5287 if isinstance(orig_element, Label):
5288 # TODO: no coverage for this block, again would be in
5289 # test_text.py where the resolve_label concept is important
5290 self._resolve_label = orig_element._label
5292 self.key = self._tq_label = self._tq_key_label = self.name
5293 self._element = element
5295 self.type = (
5296 type_api.to_instance(type_)
5297 if type_ is not None
5298 else self._element.type
5299 )
5301 self._proxies = [element]
5303 def __reduce__(self):
5304 return self.__class__, (self.name, self._element, self.type)
5306 @HasMemoized.memoized_attribute
5307 def _render_label_in_columns_clause(self):
5308 return True
5310 def _bind_param(self, operator, obj, type_=None, expanding=False):
5311 return BindParameter(
5312 None,
5313 obj,
5314 _compared_to_operator=operator,
5315 type_=type_,
5316 _compared_to_type=self.type,
5317 unique=True,
5318 expanding=expanding,
5319 )
5321 @util.memoized_property
5322 def _is_implicitly_boolean(self):
5323 return self.element._is_implicitly_boolean
5325 @HasMemoized.memoized_attribute
5326 def _allow_label_resolve(self):
5327 return self.element._allow_label_resolve
5329 @property
5330 def _order_by_label_element(self):
5331 return self
5333 def as_reference(self) -> _label_reference[_T]:
5334 """refer to this labeled expression in a clause such as GROUP BY,
5335 ORDER BY etc. as the label name itself, without expanding
5336 into the full expression.
5338 .. versionadded:: 2.1
5340 """
5341 return _label_reference(self)
5343 @HasMemoized.memoized_attribute
5344 def element(self) -> ColumnElement[_T]:
5345 return self._element.self_group(against=operators.as_)
5347 def self_group(self, against: Optional[OperatorType] = None) -> Label[_T]:
5348 return self._apply_to_inner(self._element.self_group, against=against)
5350 def _negate(self):
5351 return self._apply_to_inner(self._element._negate)
5353 def _apply_to_inner(
5354 self,
5355 fn: Callable[_PS, ColumnElement[_T]],
5356 *arg: _PS.args,
5357 **kw: _PS.kwargs,
5358 ) -> Label[_T]:
5359 sub_element = fn(*arg, **kw)
5360 if sub_element is not self._element:
5361 return Label(self.name, sub_element, type_=self.type)
5362 else:
5363 return self
5365 @property
5366 def primary_key(self): # type: ignore[override]
5367 return self.element.primary_key
5369 @property
5370 def foreign_keys(self): # type: ignore[override]
5371 return self.element.foreign_keys
5373 def _copy_internals(
5374 self,
5375 *,
5376 clone: _CloneCallableType = _clone,
5377 anonymize_labels: bool = False,
5378 **kw: Any,
5379 ) -> None:
5380 self._reset_memoizations()
5381 self._element = clone(self._element, **kw)
5382 if anonymize_labels:
5383 self.name = _anonymous_label.safe_construct(
5384 id(self), getattr(self.element, "name", "anon")
5385 )
5386 self.key = self._tq_label = self._tq_key_label = self.name
5388 @util.ro_non_memoized_property
5389 def _from_objects(self) -> List[FromClause]:
5390 return self.element._from_objects
5392 def _make_proxy(
5393 self,
5394 selectable: FromClause,
5395 *,
5396 primary_key: ColumnSet,
5397 foreign_keys: Set[KeyedColumnElement[Any]],
5398 name: Optional[str] = None,
5399 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5400 **kw: Any,
5401 ) -> typing_Tuple[str, ColumnClause[_T]]:
5402 name = self.name if not name else name
5404 key, e = self.element._make_proxy(
5405 selectable,
5406 name=name,
5407 disallow_is_literal=True,
5408 name_is_truncatable=isinstance(name, _truncated_label),
5409 compound_select_cols=compound_select_cols,
5410 primary_key=primary_key,
5411 foreign_keys=foreign_keys,
5412 )
5414 # there was a note here to remove this assertion, which was here
5415 # to determine if we later could support a use case where
5416 # the key and name of a label are separate. But I don't know what
5417 # that case was. For now, this is an unexpected case that occurs
5418 # when a label name conflicts with other columns and select()
5419 # is attempting to disambiguate an explicit label, which is not what
5420 # the user would want. See issue #6090.
5421 if key != self.name and not isinstance(self.name, _anonymous_label):
5422 raise exc.InvalidRequestError(
5423 "Label name %s is being renamed to an anonymous label due "
5424 "to disambiguation "
5425 "which is not supported right now. Please use unique names "
5426 "for explicit labels." % (self.name)
5427 )
5429 e._propagate_attrs = selectable._propagate_attrs
5430 e._proxies.append(self)
5431 if self.type is not None:
5432 e.type = self.type
5434 return self.key, e
5437class ColumnClause(
5438 roles.DDLReferredColumnRole,
5439 roles.LabeledColumnExprRole[_T],
5440 roles.StrAsPlainColumnRole,
5441 Immutable,
5442 NamedColumn[_T],
5443):
5444 """Represents a column expression from any textual string.
5446 The :class:`.ColumnClause`, a lightweight analogue to the
5447 :class:`_schema.Column` class, is typically invoked using the
5448 :func:`_expression.column` function, as in::
5450 from sqlalchemy import column
5452 id, name = column("id"), column("name")
5453 stmt = select(id, name).select_from("user")
5455 The above statement would produce SQL like:
5457 .. sourcecode:: sql
5459 SELECT id, name FROM user
5461 :class:`.ColumnClause` is the immediate superclass of the schema-specific
5462 :class:`_schema.Column` object. While the :class:`_schema.Column`
5463 class has all the
5464 same capabilities as :class:`.ColumnClause`, the :class:`.ColumnClause`
5465 class is usable by itself in those cases where behavioral requirements
5466 are limited to simple SQL expression generation. The object has none of
5467 the associations with schema-level metadata or with execution-time
5468 behavior that :class:`_schema.Column` does,
5469 so in that sense is a "lightweight"
5470 version of :class:`_schema.Column`.
5472 Full details on :class:`.ColumnClause` usage is at
5473 :func:`_expression.column`.
5475 .. seealso::
5477 :func:`_expression.column`
5479 :class:`_schema.Column`
5481 """
5483 table: Optional[FromClause]
5484 is_literal: bool
5486 __visit_name__ = "column"
5488 _traverse_internals: _TraverseInternalsType = [
5489 ("name", InternalTraversal.dp_anon_name),
5490 ("type", InternalTraversal.dp_type),
5491 ("table", InternalTraversal.dp_clauseelement),
5492 ("is_literal", InternalTraversal.dp_boolean),
5493 ]
5495 onupdate: Optional[DefaultGenerator] = None
5496 default: Optional[DefaultGenerator] = None
5497 server_default: Optional[FetchedValue] = None
5498 server_onupdate: Optional[FetchedValue] = None
5500 _is_multiparam_column = False
5502 @property
5503 def _is_star(self): # type: ignore[override]
5504 return self.is_literal and self.name == "*"
5506 def __init__(
5507 self,
5508 text: str,
5509 type_: Optional[_CovariantTypeEngineArgument[_T]] = None,
5510 is_literal: bool = False,
5511 _selectable: Optional[FromClause] = None,
5512 ):
5513 self.key = self.name = text
5514 self.table = _selectable
5516 # if type is None, we get NULLTYPE, which is our _T. But I don't
5517 # know how to get the overloads to express that correctly
5518 self.type = type_api.to_instance(type_) # type: ignore[assignment]
5520 self.is_literal = is_literal
5522 def get_children(self, *, column_tables=False, **kw):
5523 # override base get_children() to not return the Table
5524 # or selectable that is parent to this column. Traversals
5525 # expect the columns of tables and subqueries to be leaf nodes.
5526 return []
5528 @property
5529 def entity_namespace(self):
5530 if self.table is not None:
5531 return self.table.entity_namespace
5532 else:
5533 return super().entity_namespace
5535 def _clone(self, detect_subquery_cols=False, **kw):
5536 if (
5537 detect_subquery_cols
5538 and self.table is not None
5539 and self.table._is_subquery
5540 ):
5541 clone = kw.pop("clone")
5542 table = clone(self.table, **kw)
5543 new = table.c.corresponding_column(self)
5544 return new
5546 return super()._clone(**kw)
5548 @HasMemoized_ro_memoized_attribute
5549 def _from_objects(self) -> List[FromClause]:
5550 t = self.table
5551 if t is not None:
5552 return [t]
5553 else:
5554 return []
5556 @HasMemoized.memoized_attribute
5557 def _render_label_in_columns_clause(self):
5558 return self.table is not None
5560 @property
5561 def _ddl_label(self):
5562 return self._gen_tq_label(self.name, dedupe_on_key=False)
5564 def _compare_name_for_result(self, other):
5565 if (
5566 self.is_literal
5567 or self.table is None
5568 or self.table._is_textual
5569 or not hasattr(other, "proxy_set")
5570 or (
5571 isinstance(other, ColumnClause)
5572 and (
5573 other.is_literal
5574 or other.table is None
5575 or other.table._is_textual
5576 )
5577 )
5578 ):
5579 return (hasattr(other, "name") and self.name == other.name) or (
5580 hasattr(other, "_tq_label")
5581 and self._tq_label == other._tq_label
5582 )
5583 else:
5584 return other.proxy_set.intersection(self.proxy_set)
5586 def _gen_tq_label(
5587 self, name: str, dedupe_on_key: bool = True
5588 ) -> Optional[str]:
5589 """generate table-qualified label
5591 for a table-bound column this is <tablename>_<columnname>.
5593 used primarily for LABEL_STYLE_TABLENAME_PLUS_COL
5594 as well as the .columns collection on a Join object.
5596 """
5597 label: str
5598 t = self.table
5599 if self.is_literal:
5600 return None
5601 elif t is not None and is_named_from_clause(t):
5602 if has_schema_attr(t) and t.schema:
5603 label = (
5604 t.schema.replace(".", "_") + "_" + t.name + ("_" + name)
5605 )
5606 else:
5607 assert not TYPE_CHECKING or isinstance(t, NamedFromClause)
5608 label = t.name + ("_" + name)
5610 # propagate name quoting rules for labels.
5611 if is_quoted_name(name) and name.quote is not None:
5612 if is_quoted_name(label):
5613 label.quote = name.quote
5614 else:
5615 label = quoted_name(label, name.quote)
5616 elif is_quoted_name(t.name) and t.name.quote is not None:
5617 # can't get this situation to occur, so let's
5618 # assert false on it for now
5619 assert not isinstance(label, quoted_name)
5620 label = quoted_name(label, t.name.quote)
5622 if dedupe_on_key:
5623 # ensure the label name doesn't conflict with that of an
5624 # existing column. note that this implies that any Column
5625 # must **not** set up its _label before its parent table has
5626 # all of its other Column objects set up. There are several
5627 # tables in the test suite which will fail otherwise; example:
5628 # table "owner" has columns "name" and "owner_name". Therefore
5629 # column owner.name cannot use the label "owner_name", it has
5630 # to be "owner_name_1".
5631 if label in t.c:
5632 _label = label
5633 counter = 1
5634 while _label in t.c:
5635 _label = label + f"_{counter}"
5636 counter += 1
5637 label = _label
5639 return coercions.expect(roles.TruncatedLabelRole, label)
5641 else:
5642 return name
5644 def _make_proxy(
5645 self,
5646 selectable: FromClause,
5647 *,
5648 primary_key: ColumnSet,
5649 foreign_keys: Set[KeyedColumnElement[Any]],
5650 name: Optional[str] = None,
5651 key: Optional[str] = None,
5652 name_is_truncatable: bool = False,
5653 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5654 disallow_is_literal: bool = False,
5655 **kw: Any,
5656 ) -> typing_Tuple[str, ColumnClause[_T]]:
5657 # the "is_literal" flag normally should never be propagated; a proxied
5658 # column is always a SQL identifier and never the actual expression
5659 # being evaluated. however, there is a case where the "is_literal" flag
5660 # might be used to allow the given identifier to have a fixed quoting
5661 # pattern already, so maintain the flag for the proxy unless a
5662 # :class:`.Label` object is creating the proxy. See [ticket:4730].
5663 is_literal = (
5664 not disallow_is_literal
5665 and self.is_literal
5666 and (
5667 # note this does not accommodate for quoted_name differences
5668 # right now
5669 name is None
5670 or name == self.name
5671 )
5672 )
5673 c = self._constructor(
5674 (
5675 coercions.expect(roles.TruncatedLabelRole, name or self.name)
5676 if name_is_truncatable
5677 else (name or self.name)
5678 ),
5679 type_=self.type,
5680 _selectable=selectable,
5681 is_literal=is_literal,
5682 )
5683 c._propagate_attrs = selectable._propagate_attrs
5684 if name is None:
5685 c.key = self.key
5686 if compound_select_cols:
5687 c._proxies = list(compound_select_cols)
5688 else:
5689 c._proxies = [self]
5691 if selectable._is_clone_of is not None:
5692 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5693 return c.key, c
5696class TableValuedColumn(NamedColumn[_T]):
5697 __visit_name__ = "table_valued_column"
5699 _traverse_internals: _TraverseInternalsType = [
5700 ("name", InternalTraversal.dp_anon_name),
5701 ("type", InternalTraversal.dp_type),
5702 ("scalar_alias", InternalTraversal.dp_clauseelement),
5703 ]
5705 def __init__(self, scalar_alias: NamedFromClause, type_: TypeEngine[_T]):
5706 self.scalar_alias = scalar_alias
5707 self.key = self.name = scalar_alias.name
5708 self.type = type_
5710 def _copy_internals(
5711 self, clone: _CloneCallableType = _clone, **kw: Any
5712 ) -> None:
5713 self.scalar_alias = clone(self.scalar_alias, **kw)
5714 self.key = self.name = self.scalar_alias.name
5716 @util.ro_non_memoized_property
5717 def _from_objects(self) -> List[FromClause]:
5718 return [self.scalar_alias]
5721class CollationClause(ColumnElement[str]):
5722 __visit_name__ = "collation"
5724 _traverse_internals: _TraverseInternalsType = [
5725 ("collation", InternalTraversal.dp_string),
5726 ("collation_schema", InternalTraversal.dp_string),
5727 ]
5729 @classmethod
5730 @util.preload_module("sqlalchemy.sql.sqltypes")
5731 def _create_collation_expression(
5732 cls,
5733 expression: _ColumnExpressionArgument[str],
5734 collation: str,
5735 collation_schema: Optional[str] = None,
5736 ) -> BinaryExpression[str]:
5738 sqltypes = util.preloaded.sql_sqltypes
5740 expr = coercions.expect(roles.ExpressionElementRole[str], expression)
5742 if expr.type._type_affinity is sqltypes.String:
5743 collate_type = expr.type._with_collation(
5744 collation, collation_schema
5745 )
5746 else:
5747 collate_type = expr.type
5749 return BinaryExpression(
5750 expr,
5751 CollationClause(collation, collation_schema),
5752 operators.collate,
5753 type_=collate_type,
5754 )
5756 def __init__(self, collation, collation_schema=None):
5757 self.collation = collation
5758 self.collation_schema = collation_schema
5761class _IdentifiedClause(Executable, ClauseElement):
5762 __visit_name__ = "identified"
5764 def __init__(self, ident):
5765 self.ident = ident
5768class SavepointClause(_IdentifiedClause):
5769 __visit_name__ = "savepoint"
5770 inherit_cache = False
5773class RollbackToSavepointClause(_IdentifiedClause):
5774 __visit_name__ = "rollback_to_savepoint"
5775 inherit_cache = False
5778class ReleaseSavepointClause(_IdentifiedClause):
5779 __visit_name__ = "release_savepoint"
5780 inherit_cache = False
5783class quoted_name(util.MemoizedSlots, str):
5784 """Represent a SQL identifier combined with quoting preferences.
5786 :class:`.quoted_name` is a Python unicode/str subclass which
5787 represents a particular identifier name along with a
5788 ``quote`` flag. This ``quote`` flag, when set to
5789 ``True`` or ``False``, overrides automatic quoting behavior
5790 for this identifier in order to either unconditionally quote
5791 or to not quote the name. If left at its default of ``None``,
5792 quoting behavior is applied to the identifier on a per-backend basis
5793 based on an examination of the token itself.
5795 A :class:`.quoted_name` object with ``quote=True`` is also
5796 prevented from being modified in the case of a so-called
5797 "name normalize" option. Certain database backends, such as
5798 Oracle Database, Firebird, and DB2 "normalize" case-insensitive names
5799 as uppercase. The SQLAlchemy dialects for these backends
5800 convert from SQLAlchemy's lower-case-means-insensitive convention
5801 to the upper-case-means-insensitive conventions of those backends.
5802 The ``quote=True`` flag here will prevent this conversion from occurring
5803 to support an identifier that's quoted as all lower case against
5804 such a backend.
5806 The :class:`.quoted_name` object is normally created automatically
5807 when specifying the name for key schema constructs such as
5808 :class:`_schema.Table`, :class:`_schema.Column`, and others.
5809 The class can also be
5810 passed explicitly as the name to any function that receives a name which
5811 can be quoted, such as :meth:`.Inspector.has_table` with an
5812 unconditionally quoted name::
5814 from sqlalchemy import create_engine
5815 from sqlalchemy import inspect
5816 from sqlalchemy.sql import quoted_name
5818 engine = create_engine("oracle+oracledb://some_dsn")
5819 print(inspect(engine).has_table(quoted_name("some_table", True)))
5821 The above logic will run the "has table" logic against the Oracle Database
5822 backend, passing the name exactly as ``"some_table"`` without converting to
5823 upper case.
5825 A :class:`.quoted_name` object with ``quote=False`` may be passed to APIs
5826 that apply automatic quoting in order to keep the given name unquoted,
5827 such as when a PostgreSQL ``INHERITS`` option refers to a schema-qualified
5828 table name like ``my_schema.some_table``.
5830 """
5832 __slots__ = "quote", "lower", "upper"
5834 quote: Optional[bool]
5836 @overload
5837 @classmethod
5838 def construct(cls, value: str, quote: Optional[bool]) -> quoted_name: ...
5840 @overload
5841 @classmethod
5842 def construct(cls, value: None, quote: Optional[bool]) -> None: ...
5844 @classmethod
5845 def construct(
5846 cls, value: Optional[str], quote: Optional[bool]
5847 ) -> Optional[quoted_name]:
5848 if value is None:
5849 return None
5850 else:
5851 return quoted_name(value, quote)
5853 def __new__(cls, value: str, quote: Optional[bool]) -> Self:
5854 assert (
5855 value is not None
5856 ), "use quoted_name.construct() for None passthrough"
5857 if isinstance(value, cls) and (quote is None or value.quote == quote):
5858 return value
5859 self = super().__new__(cls, value)
5861 self.quote = quote
5862 return self
5864 def __reduce__(self):
5865 return quoted_name, (str(self), self.quote)
5867 def _memoized_method_lower(self):
5868 if self.quote:
5869 return self
5870 else:
5871 return str(self).lower()
5873 def _memoized_method_upper(self):
5874 if self.quote:
5875 return self
5876 else:
5877 return str(self).upper()
5880def _find_columns(clause: ClauseElement) -> Set[ColumnClause[Any]]:
5881 """locate Column objects within the given expression."""
5883 cols: Set[ColumnClause[Any]] = set()
5884 traverse(clause, {}, {"column": cols.add})
5885 return cols
5888def _type_from_args(args: Sequence[ColumnElement[_T]]) -> TypeEngine[_T]:
5889 for a in args:
5890 if not a.type._isnull:
5891 return a.type
5892 else:
5893 return type_api.NULLTYPE # type: ignore[return-value]
5896def _corresponding_column_or_error(fromclause, column, require_embedded=False):
5897 c = fromclause.corresponding_column(
5898 column, require_embedded=require_embedded
5899 )
5900 if c is None:
5901 raise exc.InvalidRequestError(
5902 "Given column '%s', attached to table '%s', "
5903 "failed to locate a corresponding column from table '%s'"
5904 % (column, getattr(column, "table", None), fromclause.description)
5905 )
5906 return c
5909class _memoized_property_but_not_nulltype(
5910 util.memoized_property["TypeEngine[_T]"]
5911):
5912 """memoized property, but dont memoize NullType"""
5914 def __get__(self, obj, cls):
5915 if obj is None:
5916 return self
5917 result = self.fget(obj)
5918 if not result._isnull:
5919 obj.__dict__[self.__name__] = result
5920 return result
5923class AnnotatedColumnElement(Annotated):
5924 _Annotated__element: ColumnElement[Any]
5926 def __init__(self, element, values):
5927 Annotated.__init__(self, element, values)
5928 for attr in (
5929 "comparator",
5930 "_proxy_key",
5931 "_tq_key_label",
5932 "_tq_label",
5933 "_non_anon_label",
5934 "type",
5935 ):
5936 self.__dict__.pop(attr, None)
5937 for attr in ("name", "key", "table"):
5938 if self.__dict__.get(attr, False) is None:
5939 self.__dict__.pop(attr)
5941 def _with_annotations(self, values):
5942 clone = super()._with_annotations(values)
5943 for attr in (
5944 "comparator",
5945 "_proxy_key",
5946 "_tq_key_label",
5947 "_tq_label",
5948 "_non_anon_label",
5949 ):
5950 clone.__dict__.pop(attr, None)
5951 return clone
5953 @util.memoized_property
5954 def name(self):
5955 """pull 'name' from parent, if not present"""
5956 return self._Annotated__element.name
5958 @_memoized_property_but_not_nulltype
5959 def type(self):
5960 """pull 'type' from parent and don't cache if null.
5962 type is routinely changed on existing columns within the
5963 mapped_column() initialization process, and "type" is also consulted
5964 during the creation of SQL expressions. Therefore it can change after
5965 it was already retrieved. At the same time we don't want annotated
5966 objects having overhead when expressions are produced, so continue
5967 to memoize, but only when we have a non-null type.
5969 """
5970 return self._Annotated__element.type
5972 @util.memoized_property
5973 def table(self):
5974 """pull 'table' from parent, if not present"""
5975 return self._Annotated__element.table
5977 @util.memoized_property
5978 def key(self):
5979 """pull 'key' from parent, if not present"""
5980 return self._Annotated__element.key
5982 @util.memoized_property
5983 def info(self) -> _InfoType:
5984 if TYPE_CHECKING:
5985 assert isinstance(self._Annotated__element, Column)
5986 return self._Annotated__element.info
5988 @util.memoized_property
5989 def _anon_name_label(self) -> str:
5990 return self._Annotated__element._anon_name_label
5993class _truncated_label(quoted_name):
5994 """A unicode subclass used to identify symbolic "
5995 "names that may require truncation."""
5997 __slots__ = ()
5999 def __new__(cls, value: str, quote: Optional[bool] = None) -> Self:
6000 quote = getattr(value, "quote", quote)
6001 # return super(_truncated_label, cls).__new__(cls, value, quote, True)
6002 return super().__new__(cls, value, quote)
6004 def __reduce__(self) -> Any:
6005 return self.__class__, (str(self), self.quote)
6007 def apply_map(self, map_: Mapping[str, Any]) -> str:
6008 return self
6011class conv(_truncated_label):
6012 """Mark a string indicating that a name has already been converted
6013 by a naming convention.
6015 This is a string subclass that indicates a name that should not be
6016 subject to any further naming conventions.
6018 E.g. when we create a :class:`.Constraint` using a naming convention
6019 as follows::
6021 m = MetaData(
6022 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
6023 )
6024 t = Table(
6025 "t", m, Column("x", Integer), CheckConstraint("x > 5", name="x5")
6026 )
6028 The name of the above constraint will be rendered as ``"ck_t_x5"``.
6029 That is, the existing name ``x5`` is used in the naming convention as the
6030 ``constraint_name`` token.
6032 In some situations, such as in migration scripts, we may be rendering
6033 the above :class:`.CheckConstraint` with a name that's already been
6034 converted. In order to make sure the name isn't double-modified, the
6035 new name is applied using the :func:`_schema.conv` marker. We can
6036 use this explicitly as follows::
6039 m = MetaData(
6040 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
6041 )
6042 t = Table(
6043 "t",
6044 m,
6045 Column("x", Integer),
6046 CheckConstraint("x > 5", name=conv("ck_t_x5")),
6047 )
6049 Where above, the :func:`_schema.conv` marker indicates that the constraint
6050 name here is final, and the name will render as ``"ck_t_x5"`` and not
6051 ``"ck_t_ck_t_x5"``
6053 .. seealso::
6055 :ref:`constraint_naming_conventions`
6057 """
6059 __slots__ = ()
6062# for backwards compatibility in case
6063# someone is re-implementing the
6064# _truncated_identifier() sequence in a custom
6065# compiler
6066_generated_label = _truncated_label
6067_anonymous_label_escape = re.compile(r"[%\(\) \$]+")
6068# for bind parameter keys, additionally escape the characters that
6069# SQLCompiler.bindname_escape_characters would otherwise escape only at
6070# compile time, after the uniquifying counter has already been applied.
6071# escaping them up front is what allows names like "a.b" and "a_b" to be
6072# disambiguated as "a_b_1" / "a_b_2" rather than colliding. see #13534
6073_bind_key_escape = re.compile(r"[%\(\) \$\.\[\]:]+")
6076class _anonymous_label(_truncated_label):
6077 """A unicode subclass used to identify anonymously
6078 generated names."""
6080 __slots__ = ()
6082 @classmethod
6083 def safe_construct_with_key(
6084 cls, seed: int | str, body: str, sanitize_key: bool = False
6085 ) -> typing_Tuple[_anonymous_label, str]:
6086 # need to escape chars that interfere with format
6087 # strings in any case, issue #8724
6088 if sanitize_key:
6089 # sanitize_key is an extra step used by BindParameter, which
6090 # also escapes the characters that would otherwise be escaped
6091 # only at compile time; issue #13534
6092 body = _bind_key_escape.sub("_", body).strip("_")
6093 else:
6094 body = _anonymous_label_escape.sub("_", body)
6096 key = f"{seed} {body.replace('%', '%%')}"
6097 label = _anonymous_label(f"%({key})s")
6098 return label, key
6100 @classmethod
6101 def safe_construct(
6102 cls, seed: int | str, body: str, sanitize_key: bool = False
6103 ) -> _anonymous_label:
6104 # need to escape chars that interfere with format
6105 # strings in any case, issue #8724
6106 if sanitize_key:
6107 # sanitize_key is an extra step used by BindParameter, which
6108 # also escapes the characters that would otherwise be escaped
6109 # only at compile time; issue #13534
6110 body = _bind_key_escape.sub("_", body).strip("_")
6111 else:
6112 body = _anonymous_label_escape.sub("_", body)
6114 return _anonymous_label(f"%({seed} {body.replace('%', '%%')})s")
6116 def __add__(self, other: str) -> _anonymous_label:
6117 if "%" in other and not isinstance(other, _anonymous_label):
6118 other = str(other).replace("%", "%%")
6119 else:
6120 other = str(other)
6122 return _anonymous_label(
6123 quoted_name(
6124 str.__add__(self, other),
6125 self.quote,
6126 )
6127 )
6129 def __radd__(self, other: str) -> _anonymous_label:
6130 if "%" in other and not isinstance(other, _anonymous_label):
6131 other = str(other).replace("%", "%%")
6132 else:
6133 other = str(other)
6135 return _anonymous_label(
6136 quoted_name(
6137 str.__add__(other, self),
6138 self.quote,
6139 )
6140 )
6142 def apply_map(self, map_: Mapping[str, Any]) -> str:
6143 if self.quote is not None:
6144 # preserve quoting only if necessary
6145 return quoted_name(self % map_, self.quote)
6146 else:
6147 # else skip the constructor call
6148 return self % map_