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 existing_bps: List[BindParameter[Any]] = []
4540 ck = self.element._gen_cache_key(anon_map, existing_bps)
4542 bindparams.extend(
4543 (
4544 bp._with_value(
4545 self.translate[bp.key], maintain_key=True, required=False
4546 )
4547 if bp.key in self.translate
4548 else bp
4549 )
4550 for bp in existing_bps
4551 )
4553 # ck derives from _gen_cache_key, a compiled function in
4554 # _cache_key_cy that mypy sees as untyped
4555 return ck # type: ignore[no-any-return]
4557 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4558 # group the element we wrap the same way it would group on its own.
4559 # see #13753
4560 grouped = self.element.self_group(against=against)
4561 if grouped is self.element:
4562 return self
4564 new = self._clone()
4565 new.element = grouped
4566 return new
4568 def _negate(self) -> ColumnElement[Any]:
4569 # negate the element we wrap the same way it would negate on its
4570 # own, e.g. NOT (a OR b) rather than NOT a OR b. see #13753
4571 new = self._clone()
4572 new.element = self.element._negate()
4573 new.type = new.element.type
4574 return new
4576 @util.ro_non_memoized_property
4577 def _from_objects(self) -> List[FromClause]:
4578 return self.element._from_objects
4580 @util.non_memoized_property
4581 def _is_implicitly_boolean(self) -> bool:
4582 return self.element._is_implicitly_boolean
4584 @util.non_memoized_property
4585 def _proxies(self) -> List[ColumnElement[Any]]:
4586 return [self.element]
4588 @util.non_memoized_property
4589 def _tq_label(self) -> Optional[str]:
4590 return (
4591 getattr(self.element, "_tq_label", None) or self._anon_name_label
4592 )
4595_FrameIntTuple = tuple[int | None, int | None]
4598class Over(ColumnElement[_T]):
4599 """Represent an OVER clause.
4601 This is a special operator against a so-called
4602 "window" function, as well as any aggregate function,
4603 which produces results relative to the result set
4604 itself. Most modern SQL backends now support window functions.
4606 """
4608 __visit_name__ = "over"
4610 _traverse_internals: _TraverseInternalsType = [
4611 ("element", InternalTraversal.dp_clauseelement),
4612 ("order_by", InternalTraversal.dp_clauseelement),
4613 ("partition_by", InternalTraversal.dp_clauseelement),
4614 ("range_", InternalTraversal.dp_clauseelement),
4615 ("rows", InternalTraversal.dp_clauseelement),
4616 ("groups", InternalTraversal.dp_clauseelement),
4617 ("exclude", InternalTraversal.dp_string),
4618 ]
4620 order_by: Optional[ClauseList] = None
4621 partition_by: Optional[ClauseList] = None
4623 element: ColumnElement[_T]
4624 """The underlying expression object to which this :class:`.Over`
4625 object refers."""
4627 range_: FrameClause | None
4628 rows: FrameClause | None
4629 groups: FrameClause | None
4630 exclude: str | None
4632 def __init__(
4633 self,
4634 element: ColumnElement[_T],
4635 partition_by: Optional[_ByArgument] = None,
4636 order_by: Optional[_ByArgument] = None,
4637 range_: _FrameIntTuple | FrameClause | None = None,
4638 rows: _FrameIntTuple | FrameClause | None = None,
4639 groups: _FrameIntTuple | FrameClause | None = None,
4640 exclude: str | None = None,
4641 ):
4642 self.element = element
4643 if order_by is not None:
4644 self.order_by = ClauseList(
4645 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4646 )
4647 if partition_by is not None:
4648 self.partition_by = ClauseList(
4649 *util.to_list(partition_by),
4650 _literal_as_text_role=roles.ByOfRole,
4651 )
4653 if sum(item is not None for item in (range_, rows, groups)) > 1:
4654 raise exc.ArgumentError(
4655 "only one of 'rows', 'range_', or 'groups' may be provided"
4656 )
4657 else:
4658 self.range_ = FrameClause._parse(range_, coerce_int=False)
4659 self.rows = FrameClause._parse(rows, coerce_int=True)
4660 self.groups = FrameClause._parse(groups, coerce_int=True)
4662 self.exclude = exclude
4664 if exclude is not None and (
4665 range_ is None and rows is None and groups is None
4666 ):
4667 raise exc.ArgumentError(
4668 "'exclude' requires that one of 'rows', "
4669 "'range_', or 'groups' is also specified"
4670 )
4672 if not TYPE_CHECKING:
4674 @util.memoized_property
4675 def type(self) -> TypeEngine[_T]: # noqa: A001
4676 return self.element.type
4678 @util.ro_non_memoized_property
4679 def _from_objects(self) -> List[FromClause]:
4680 return list(
4681 itertools.chain(
4682 *[
4683 c._from_objects
4684 for c in (self.element, self.partition_by, self.order_by)
4685 if c is not None
4686 ]
4687 )
4688 )
4691class FrameClauseType(Enum):
4692 """Frame clause type enum for FrameClause lower_type and upper_type.
4694 .. versionadded:: 2.1
4696 """
4698 UNBOUNDED = 0
4699 """Produces an "UNBOUNDED PRECEDING" or "UNBOUNDED FOLLOWING" frame
4700 clause depending on the position.
4701 Requires a ``None`` value for the corresponding bound value.
4702 """
4703 CURRENT = 1
4704 """Produces a "CURRENT ROW" frame clause.
4705 Requires a ``None`` value for the corresponding bound value.
4706 """
4707 PRECEDING = 2
4708 """Produces a "PRECEDING" frame clause."""
4709 FOLLOWING = 3
4710 """Produces a "FOLLOWING" frame clause."""
4713_require_none = (
4714 FrameClauseType.CURRENT,
4715 FrameClauseType.UNBOUNDED,
4716)
4719class FrameClause(ClauseElement):
4720 """Indicate the 'rows' 'range' or 'group' field of a window function,
4721 e.g. using :class:`.Over`.
4723 .. versionadded:: 2.1
4725 """
4727 __visit_name__ = "frame_clause"
4729 _traverse_internals: _TraverseInternalsType = [
4730 ("lower_bind", InternalTraversal.dp_clauseelement),
4731 ("upper_bind", InternalTraversal.dp_clauseelement),
4732 ("lower_type", InternalTraversal.dp_plain_obj),
4733 ("upper_type", InternalTraversal.dp_plain_obj),
4734 ]
4736 def __init__(
4737 self,
4738 start: Any,
4739 end: Any,
4740 start_frame_type: FrameClauseType,
4741 end_frame_type: FrameClauseType,
4742 _validate: bool = True,
4743 ) -> None:
4744 """Creates a new FrameClause specifying the bounds of a window frame.
4746 :param start: The start value.
4747 :param end: The end value.
4748 :param start_frame_type: The :class:`FrameClauseType` for the
4749 start value.
4750 :param end_frame_type: The :class:`FrameClauseType` for the end value.
4751 """
4752 self.lower_bind = self._as_literal(start)
4753 self.upper_bind = self._as_literal(end)
4754 self.lower_type = FrameClauseType(start_frame_type)
4755 self.upper_type = FrameClauseType(end_frame_type)
4756 if _validate:
4757 if (
4758 self.lower_type in _require_none
4759 and self.lower_bind is not None
4760 ):
4761 raise exc.ArgumentError(
4762 "Cannot specify a value for start with frame type "
4763 f"{self.lower_type.name}"
4764 )
4765 if (
4766 self.upper_type in _require_none
4767 and self.upper_bind is not None
4768 ):
4769 raise exc.ArgumentError(
4770 "Cannot specify a value for end with frame type "
4771 f"{self.upper_type.name}"
4772 )
4774 @classmethod
4775 def _as_literal(cls, value: Any) -> BindParameter[Any] | None:
4776 if value is None:
4777 return None
4778 elif isinstance(value, int):
4779 return literal(value, type_api.INTEGERTYPE)
4780 elif isinstance(value, BindParameter):
4781 return value
4782 else:
4783 return literal(value) # let the default type resolution occur
4785 @classmethod
4786 def _handle_int(
4787 cls, value: Any | None, coerce_int: bool
4788 ) -> tuple[int | None, FrameClauseType]:
4789 if value is None:
4790 return None, FrameClauseType.UNBOUNDED
4792 if coerce_int:
4793 try:
4794 integer = int(value)
4795 except ValueError as err:
4796 raise exc.ArgumentError(
4797 "Integer or None expected for values in rows/groups frame"
4798 ) from err
4799 elif not isinstance(value, int):
4800 raise exc.ArgumentError(
4801 "When using a tuple to specify a range only integer or none "
4802 "values are allowed in the range frame. To specify a "
4803 "different type use the FrameClause directly."
4804 )
4805 else:
4806 integer = value
4807 if integer == 0:
4808 return None, FrameClauseType.CURRENT
4809 elif integer < 0:
4810 return abs(integer), FrameClauseType.PRECEDING
4811 else:
4812 return integer, FrameClauseType.FOLLOWING
4814 @classmethod
4815 def _parse(
4816 cls,
4817 range_: _FrameIntTuple | FrameClause | None,
4818 coerce_int: bool,
4819 ) -> FrameClause | None:
4820 if range_ is None or isinstance(range_, FrameClause):
4821 return range_
4823 try:
4824 r0, r1 = range_
4825 except (ValueError, TypeError) as ve:
4826 raise exc.ArgumentError(
4827 "2-tuple expected for range/rows/groups"
4828 ) from ve
4830 l_b, l_t = cls._handle_int(r0, coerce_int)
4831 u_b, u_t = cls._handle_int(r1, coerce_int)
4833 return FrameClause(
4834 start=l_b,
4835 end=u_b,
4836 start_frame_type=l_t,
4837 end_frame_type=u_t,
4838 _validate=False,
4839 )
4842class AggregateOrderBy(WrapsColumnExpression[_T]):
4843 """Represent an aggregate ORDER BY expression.
4845 This is a special operator against aggregate functions such as
4846 ``array_agg()``, ``json_arrayagg()`` ``string_agg()``, etc. that provides
4847 for an ORDER BY expression, using a syntax that's compatible with
4848 the backend.
4850 :class:`.AggregateOrderBy` is a generalized version of the
4851 :class:`.WithinGroup` construct, the latter of which always provides a
4852 "WITHIN GROUP (ORDER BY ...)" expression. :class:`.AggregateOrderBy` will
4853 also compile to "WITHIN GROUP (ORDER BY ...)" on backends such as Oracle
4854 and SQL Server that don't have another style of aggregate function
4855 ordering.
4857 .. versionadded:: 2.1
4860 """
4862 __visit_name__ = "aggregateorderby"
4864 _traverse_internals: _TraverseInternalsType = [
4865 ("element", InternalTraversal.dp_clauseelement),
4866 ("order_by", InternalTraversal.dp_clauseelement),
4867 ]
4869 order_by: ClauseList
4871 def __init__(
4872 self,
4873 element: Union[FunctionElement[_T], FunctionFilter[_T]],
4874 *order_by: _ColumnExpressionArgument[Any],
4875 ):
4876 self.element = element
4877 if not order_by:
4878 raise TypeError("at least one ORDER BY element is required")
4879 self.order_by = ClauseList(
4880 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4881 )
4883 if not TYPE_CHECKING:
4885 @util.memoized_property
4886 def type(self) -> TypeEngine[_T]: # noqa: A001
4887 return self.element.type
4889 @property
4890 def wrapped_column_expression(self) -> ColumnElement[_T]:
4891 return self.element
4893 def __reduce__(self):
4894 return self.__class__, (self.element,) + (
4895 tuple(self.order_by) if self.order_by is not None else ()
4896 )
4898 def over(
4899 self,
4900 *,
4901 partition_by: _ByArgument | None = None,
4902 order_by: _ByArgument | None = None,
4903 rows: _FrameIntTuple | FrameClause | None = None,
4904 range_: _FrameIntTuple | FrameClause | None = None,
4905 groups: _FrameIntTuple | FrameClause | None = None,
4906 exclude: str | None = None,
4907 ) -> Over[_T]:
4908 """Produce an OVER clause against this :class:`.WithinGroup`
4909 construct.
4911 This function has the same signature as that of
4912 :meth:`.FunctionElement.over`.
4914 """
4915 return Over(
4916 self,
4917 partition_by=partition_by,
4918 order_by=order_by,
4919 range_=range_,
4920 rows=rows,
4921 groups=groups,
4922 exclude=exclude,
4923 )
4925 @overload
4926 def filter(self) -> Self: ...
4928 @overload
4929 def filter(
4930 self,
4931 __criterion0: _ColumnExpressionArgument[bool],
4932 *criterion: _ColumnExpressionArgument[bool],
4933 ) -> FunctionFilter[_T]: ...
4935 def filter(
4936 self, *criterion: _ColumnExpressionArgument[bool]
4937 ) -> Union[Self, FunctionFilter[_T]]:
4938 """Produce a FILTER clause against this function."""
4939 if not criterion:
4940 return self
4941 return FunctionFilter(self, *criterion)
4943 @util.ro_non_memoized_property
4944 def _from_objects(self) -> List[FromClause]:
4945 return list(
4946 itertools.chain(
4947 *[
4948 c._from_objects
4949 for c in (self.element, self.order_by)
4950 if c is not None
4951 ]
4952 )
4953 )
4956class WithinGroup(AggregateOrderBy[_T]):
4957 """Represent a WITHIN GROUP (ORDER BY) clause.
4959 This is a special operator against so-called
4960 "ordered set aggregate" and "hypothetical
4961 set aggregate" functions, including ``percentile_cont()``,
4962 ``rank()``, ``dense_rank()``, etc.
4964 It's supported only by certain database backends, such as PostgreSQL,
4965 Oracle Database and MS SQL Server.
4967 The :class:`.WithinGroup` construct extracts its type from the
4968 method :meth:`.FunctionElement.within_group_type`. If this returns
4969 ``None``, the function's ``.type`` is used.
4971 """
4973 __visit_name__ = "withingroup"
4974 inherit_cache = True
4976 if not TYPE_CHECKING:
4978 @util.memoized_property
4979 def type(self) -> TypeEngine[_T]: # noqa: A001
4980 wgt = self.element.within_group_type(self)
4981 if wgt is not None:
4982 return wgt
4983 else:
4984 return self.element.type
4987class FunctionFilter(Generative, ColumnElement[_T]):
4988 """Represent a function FILTER clause.
4990 This is a special operator against aggregate and window functions,
4991 which controls which rows are passed to it.
4992 It's supported only by certain database backends.
4994 Invocation of :class:`.FunctionFilter` is via
4995 :meth:`.FunctionElement.filter`::
4997 func.count(1).filter(True)
4999 .. seealso::
5001 :meth:`.FunctionElement.filter`
5003 """
5005 __visit_name__ = "funcfilter"
5007 _traverse_internals: _TraverseInternalsType = [
5008 ("func", InternalTraversal.dp_clauseelement),
5009 ("criterion", InternalTraversal.dp_clauseelement),
5010 ]
5012 criterion: Optional[ColumnElement[bool]] = None
5014 def __init__(
5015 self,
5016 func: Union[FunctionElement[_T], AggregateOrderBy[_T]],
5017 *criterion: _ColumnExpressionArgument[bool],
5018 ):
5019 self.func = func
5020 self.filter.non_generative(self, *criterion) # type: ignore[attr-defined] # noqa: E501
5022 @_generative
5023 def filter(self, *criterion: _ColumnExpressionArgument[bool]) -> Self:
5024 """Produce an additional FILTER against the function.
5026 This method adds additional criteria to the initial criteria
5027 set up by :meth:`.FunctionElement.filter`.
5029 Multiple criteria are joined together at SQL render time
5030 via ``AND``.
5033 """
5035 for crit in list(criterion):
5036 crit = coercions.expect(roles.WhereHavingRole, crit)
5038 if self.criterion is not None:
5039 self.criterion = self.criterion & crit
5040 else:
5041 self.criterion = crit
5043 return self
5045 def over(
5046 self,
5047 partition_by: _ByArgument | None = None,
5048 order_by: _ByArgument | None = None,
5049 range_: _FrameIntTuple | FrameClause | None = None,
5050 rows: _FrameIntTuple | FrameClause | None = None,
5051 groups: _FrameIntTuple | FrameClause | None = None,
5052 exclude: str | None = None,
5053 ) -> Over[_T]:
5054 """Produce an OVER clause against this filtered function.
5056 Used against aggregate or so-called "window" functions,
5057 for database backends that support window functions.
5059 The expression::
5061 func.rank().filter(MyClass.y > 5).over(order_by="x")
5063 is shorthand for::
5065 from sqlalchemy import over, funcfilter
5067 over(funcfilter(func.rank(), MyClass.y > 5), order_by="x")
5069 See :func:`_expression.over` for a full description.
5071 """
5072 return Over(
5073 self,
5074 partition_by=partition_by,
5075 order_by=order_by,
5076 range_=range_,
5077 rows=rows,
5078 groups=groups,
5079 exclude=exclude,
5080 )
5082 def within_group(
5083 self, *order_by: _ColumnExpressionArgument[Any]
5084 ) -> WithinGroup[_T]:
5085 """Produce a WITHIN GROUP (ORDER BY expr) clause against
5086 this function.
5087 """
5088 return WithinGroup(self, *order_by)
5090 def within_group_type(
5091 self, within_group: WithinGroup[_T]
5092 ) -> Optional[TypeEngine[_T]]:
5093 return None
5095 def self_group(
5096 self, against: Optional[OperatorType] = None
5097 ) -> Union[Self, Grouping[_T]]:
5098 if operators.is_precedent(operators.filter_op, against):
5099 return Grouping(self)
5100 else:
5101 return self
5103 if not TYPE_CHECKING:
5105 @util.memoized_property
5106 def type(self) -> TypeEngine[_T]: # noqa: A001
5107 return self.func.type
5109 @util.ro_non_memoized_property
5110 def _from_objects(self) -> List[FromClause]:
5111 return list(
5112 itertools.chain(
5113 *[
5114 c._from_objects
5115 for c in (self.func, self.criterion)
5116 if c is not None
5117 ]
5118 )
5119 )
5122class NamedColumn(KeyedColumnElement[_T]):
5123 is_literal = False
5124 table: Optional[FromClause] = None
5125 name: str
5126 key: str
5128 def _compare_name_for_result(self, other):
5129 return (hasattr(other, "name") and self.name == other.name) or (
5130 hasattr(other, "_label") and self._label == other._label
5131 )
5133 @util.ro_memoized_property
5134 def description(self) -> str:
5135 return self.name
5137 @HasMemoized.memoized_attribute
5138 def _tq_key_label(self) -> Optional[str]:
5139 """table qualified label based on column key.
5141 for table-bound columns this is <tablename>_<column key/proxy key>;
5143 all other expressions it resolves to key/proxy key.
5145 """
5146 proxy_key = self._proxy_key
5147 if proxy_key and proxy_key != self.name:
5148 return self._gen_tq_label(proxy_key)
5149 else:
5150 return self._tq_label
5152 @HasMemoized.memoized_attribute
5153 def _tq_label(self) -> Optional[str]:
5154 """table qualified label based on column name.
5156 for table-bound columns this is <tablename>_<columnname>; all other
5157 expressions it resolves to .name.
5159 """
5160 return self._gen_tq_label(self.name)
5162 @HasMemoized.memoized_attribute
5163 def _render_label_in_columns_clause(self):
5164 return True
5166 @HasMemoized.memoized_attribute
5167 def _non_anon_label(self):
5168 return self.name
5170 def _gen_tq_label(
5171 self, name: str, dedupe_on_key: bool = True
5172 ) -> Optional[str]:
5173 return name
5175 def _bind_param(
5176 self,
5177 operator: OperatorType,
5178 obj: Any,
5179 type_: Optional[TypeEngine[_T]] = None,
5180 expanding: bool = False,
5181 ) -> BindParameter[_T]:
5182 return BindParameter(
5183 self.key,
5184 obj,
5185 _compared_to_operator=operator,
5186 _compared_to_type=self.type,
5187 type_=type_,
5188 unique=True,
5189 expanding=expanding,
5190 )
5192 def _make_proxy(
5193 self,
5194 selectable: FromClause,
5195 *,
5196 primary_key: ColumnSet,
5197 foreign_keys: Set[KeyedColumnElement[Any]],
5198 name: Optional[str] = None,
5199 key: Optional[str] = None,
5200 name_is_truncatable: bool = False,
5201 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5202 disallow_is_literal: bool = False,
5203 **kw: Any,
5204 ) -> typing_Tuple[str, ColumnClause[_T]]:
5205 c = ColumnClause(
5206 (
5207 coercions.expect(roles.TruncatedLabelRole, name or self.name)
5208 if name_is_truncatable
5209 else (name or self.name)
5210 ),
5211 type_=self.type,
5212 _selectable=selectable,
5213 is_literal=False,
5214 )
5216 c._propagate_attrs = selectable._propagate_attrs
5217 if name is None:
5218 c.key = self.key
5219 if compound_select_cols:
5220 c._proxies = list(compound_select_cols)
5221 else:
5222 c._proxies = [self]
5224 if selectable._is_clone_of is not None:
5225 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5226 return c.key, c
5229_PS = ParamSpec("_PS")
5232class Label(roles.LabeledColumnExprRole[_T], NamedColumn[_T]):
5233 """Represents a column label (AS).
5235 Represent a label, as typically applied to any column-level
5236 element using the ``AS`` sql keyword.
5238 """
5240 __visit_name__ = "label"
5242 _traverse_internals: _TraverseInternalsType = [
5243 ("name", InternalTraversal.dp_anon_name),
5244 ("type", InternalTraversal.dp_type),
5245 ("_element", InternalTraversal.dp_clauseelement),
5246 ]
5248 _cache_key_traversal = [
5249 ("name", InternalTraversal.dp_anon_name),
5250 ("_element", InternalTraversal.dp_clauseelement),
5251 ]
5253 _element: ColumnElement[_T]
5254 name: str
5256 def __init__(
5257 self,
5258 name: Optional[str],
5259 element: _ColumnExpressionArgument[_T],
5260 type_: Optional[_CovariantTypeEngineArgument[_T]] = None,
5261 ):
5262 orig_element = element
5263 element = coercions.expect(
5264 roles.ExpressionElementRole,
5265 element,
5266 apply_propagate_attrs=self,
5267 )
5268 while isinstance(element, Label):
5269 # TODO: this is only covered in test_text.py, but nothing
5270 # fails if it's removed. determine rationale
5271 element = element.element
5273 if name:
5274 self.name = name
5275 else:
5276 self.name = _anonymous_label.safe_construct(
5277 id(self), getattr(element, "name", "anon")
5278 )
5279 if isinstance(orig_element, Label):
5280 # TODO: no coverage for this block, again would be in
5281 # test_text.py where the resolve_label concept is important
5282 self._resolve_label = orig_element._label
5284 self.key = self._tq_label = self._tq_key_label = self.name
5285 self._element = element
5287 self.type = (
5288 type_api.to_instance(type_)
5289 if type_ is not None
5290 else self._element.type
5291 )
5293 self._proxies = [element]
5295 def __reduce__(self):
5296 return self.__class__, (self.name, self._element, self.type)
5298 @HasMemoized.memoized_attribute
5299 def _render_label_in_columns_clause(self):
5300 return True
5302 def _bind_param(self, operator, obj, type_=None, expanding=False):
5303 return BindParameter(
5304 None,
5305 obj,
5306 _compared_to_operator=operator,
5307 type_=type_,
5308 _compared_to_type=self.type,
5309 unique=True,
5310 expanding=expanding,
5311 )
5313 @util.memoized_property
5314 def _is_implicitly_boolean(self):
5315 return self.element._is_implicitly_boolean
5317 @HasMemoized.memoized_attribute
5318 def _allow_label_resolve(self):
5319 return self.element._allow_label_resolve
5321 @property
5322 def _order_by_label_element(self):
5323 return self
5325 def as_reference(self) -> _label_reference[_T]:
5326 """refer to this labeled expression in a clause such as GROUP BY,
5327 ORDER BY etc. as the label name itself, without expanding
5328 into the full expression.
5330 .. versionadded:: 2.1
5332 """
5333 return _label_reference(self)
5335 @HasMemoized.memoized_attribute
5336 def element(self) -> ColumnElement[_T]:
5337 return self._element.self_group(against=operators.as_)
5339 def self_group(self, against: Optional[OperatorType] = None) -> Label[_T]:
5340 return self._apply_to_inner(self._element.self_group, against=against)
5342 def _negate(self):
5343 return self._apply_to_inner(self._element._negate)
5345 def _apply_to_inner(
5346 self,
5347 fn: Callable[_PS, ColumnElement[_T]],
5348 *arg: _PS.args,
5349 **kw: _PS.kwargs,
5350 ) -> Label[_T]:
5351 sub_element = fn(*arg, **kw)
5352 if sub_element is not self._element:
5353 return Label(self.name, sub_element, type_=self.type)
5354 else:
5355 return self
5357 @property
5358 def primary_key(self): # type: ignore[override]
5359 return self.element.primary_key
5361 @property
5362 def foreign_keys(self): # type: ignore[override]
5363 return self.element.foreign_keys
5365 def _copy_internals(
5366 self,
5367 *,
5368 clone: _CloneCallableType = _clone,
5369 anonymize_labels: bool = False,
5370 **kw: Any,
5371 ) -> None:
5372 self._reset_memoizations()
5373 self._element = clone(self._element, **kw)
5374 if anonymize_labels:
5375 self.name = _anonymous_label.safe_construct(
5376 id(self), getattr(self.element, "name", "anon")
5377 )
5378 self.key = self._tq_label = self._tq_key_label = self.name
5380 @util.ro_non_memoized_property
5381 def _from_objects(self) -> List[FromClause]:
5382 return self.element._from_objects
5384 def _make_proxy(
5385 self,
5386 selectable: FromClause,
5387 *,
5388 primary_key: ColumnSet,
5389 foreign_keys: Set[KeyedColumnElement[Any]],
5390 name: Optional[str] = None,
5391 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5392 **kw: Any,
5393 ) -> typing_Tuple[str, ColumnClause[_T]]:
5394 name = self.name if not name else name
5396 key, e = self.element._make_proxy(
5397 selectable,
5398 name=name,
5399 disallow_is_literal=True,
5400 name_is_truncatable=isinstance(name, _truncated_label),
5401 compound_select_cols=compound_select_cols,
5402 primary_key=primary_key,
5403 foreign_keys=foreign_keys,
5404 )
5406 # there was a note here to remove this assertion, which was here
5407 # to determine if we later could support a use case where
5408 # the key and name of a label are separate. But I don't know what
5409 # that case was. For now, this is an unexpected case that occurs
5410 # when a label name conflicts with other columns and select()
5411 # is attempting to disambiguate an explicit label, which is not what
5412 # the user would want. See issue #6090.
5413 if key != self.name and not isinstance(self.name, _anonymous_label):
5414 raise exc.InvalidRequestError(
5415 "Label name %s is being renamed to an anonymous label due "
5416 "to disambiguation "
5417 "which is not supported right now. Please use unique names "
5418 "for explicit labels." % (self.name)
5419 )
5421 e._propagate_attrs = selectable._propagate_attrs
5422 e._proxies.append(self)
5423 if self.type is not None:
5424 e.type = self.type
5426 return self.key, e
5429class ColumnClause(
5430 roles.DDLReferredColumnRole,
5431 roles.LabeledColumnExprRole[_T],
5432 roles.StrAsPlainColumnRole,
5433 Immutable,
5434 NamedColumn[_T],
5435):
5436 """Represents a column expression from any textual string.
5438 The :class:`.ColumnClause`, a lightweight analogue to the
5439 :class:`_schema.Column` class, is typically invoked using the
5440 :func:`_expression.column` function, as in::
5442 from sqlalchemy import column
5444 id, name = column("id"), column("name")
5445 stmt = select(id, name).select_from("user")
5447 The above statement would produce SQL like:
5449 .. sourcecode:: sql
5451 SELECT id, name FROM user
5453 :class:`.ColumnClause` is the immediate superclass of the schema-specific
5454 :class:`_schema.Column` object. While the :class:`_schema.Column`
5455 class has all the
5456 same capabilities as :class:`.ColumnClause`, the :class:`.ColumnClause`
5457 class is usable by itself in those cases where behavioral requirements
5458 are limited to simple SQL expression generation. The object has none of
5459 the associations with schema-level metadata or with execution-time
5460 behavior that :class:`_schema.Column` does,
5461 so in that sense is a "lightweight"
5462 version of :class:`_schema.Column`.
5464 Full details on :class:`.ColumnClause` usage is at
5465 :func:`_expression.column`.
5467 .. seealso::
5469 :func:`_expression.column`
5471 :class:`_schema.Column`
5473 """
5475 table: Optional[FromClause]
5476 is_literal: bool
5478 __visit_name__ = "column"
5480 _traverse_internals: _TraverseInternalsType = [
5481 ("name", InternalTraversal.dp_anon_name),
5482 ("type", InternalTraversal.dp_type),
5483 ("table", InternalTraversal.dp_clauseelement),
5484 ("is_literal", InternalTraversal.dp_boolean),
5485 ]
5487 onupdate: Optional[DefaultGenerator] = None
5488 default: Optional[DefaultGenerator] = None
5489 server_default: Optional[FetchedValue] = None
5490 server_onupdate: Optional[FetchedValue] = None
5492 _is_multiparam_column = False
5494 @property
5495 def _is_star(self): # type: ignore[override]
5496 return self.is_literal and self.name == "*"
5498 def __init__(
5499 self,
5500 text: str,
5501 type_: Optional[_CovariantTypeEngineArgument[_T]] = None,
5502 is_literal: bool = False,
5503 _selectable: Optional[FromClause] = None,
5504 ):
5505 self.key = self.name = text
5506 self.table = _selectable
5508 # if type is None, we get NULLTYPE, which is our _T. But I don't
5509 # know how to get the overloads to express that correctly
5510 self.type = type_api.to_instance(type_) # type: ignore[assignment]
5512 self.is_literal = is_literal
5514 def get_children(self, *, column_tables=False, **kw):
5515 # override base get_children() to not return the Table
5516 # or selectable that is parent to this column. Traversals
5517 # expect the columns of tables and subqueries to be leaf nodes.
5518 return []
5520 @property
5521 def entity_namespace(self):
5522 if self.table is not None:
5523 return self.table.entity_namespace
5524 else:
5525 return super().entity_namespace
5527 def _clone(self, detect_subquery_cols=False, **kw):
5528 if (
5529 detect_subquery_cols
5530 and self.table is not None
5531 and self.table._is_subquery
5532 ):
5533 clone = kw.pop("clone")
5534 table = clone(self.table, **kw)
5535 new = table.c.corresponding_column(self)
5536 return new
5538 return super()._clone(**kw)
5540 @HasMemoized_ro_memoized_attribute
5541 def _from_objects(self) -> List[FromClause]:
5542 t = self.table
5543 if t is not None:
5544 return [t]
5545 else:
5546 return []
5548 @HasMemoized.memoized_attribute
5549 def _render_label_in_columns_clause(self):
5550 return self.table is not None
5552 @property
5553 def _ddl_label(self):
5554 return self._gen_tq_label(self.name, dedupe_on_key=False)
5556 def _compare_name_for_result(self, other):
5557 if (
5558 self.is_literal
5559 or self.table is None
5560 or self.table._is_textual
5561 or not hasattr(other, "proxy_set")
5562 or (
5563 isinstance(other, ColumnClause)
5564 and (
5565 other.is_literal
5566 or other.table is None
5567 or other.table._is_textual
5568 )
5569 )
5570 ):
5571 return (hasattr(other, "name") and self.name == other.name) or (
5572 hasattr(other, "_tq_label")
5573 and self._tq_label == other._tq_label
5574 )
5575 else:
5576 return other.proxy_set.intersection(self.proxy_set)
5578 def _gen_tq_label(
5579 self, name: str, dedupe_on_key: bool = True
5580 ) -> Optional[str]:
5581 """generate table-qualified label
5583 for a table-bound column this is <tablename>_<columnname>.
5585 used primarily for LABEL_STYLE_TABLENAME_PLUS_COL
5586 as well as the .columns collection on a Join object.
5588 """
5589 label: str
5590 t = self.table
5591 if self.is_literal:
5592 return None
5593 elif t is not None and is_named_from_clause(t):
5594 if has_schema_attr(t) and t.schema:
5595 label = (
5596 t.schema.replace(".", "_") + "_" + t.name + ("_" + name)
5597 )
5598 else:
5599 assert not TYPE_CHECKING or isinstance(t, NamedFromClause)
5600 label = t.name + ("_" + name)
5602 # propagate name quoting rules for labels.
5603 if is_quoted_name(name) and name.quote is not None:
5604 if is_quoted_name(label):
5605 label.quote = name.quote
5606 else:
5607 label = quoted_name(label, name.quote)
5608 elif is_quoted_name(t.name) and t.name.quote is not None:
5609 # can't get this situation to occur, so let's
5610 # assert false on it for now
5611 assert not isinstance(label, quoted_name)
5612 label = quoted_name(label, t.name.quote)
5614 if dedupe_on_key:
5615 # ensure the label name doesn't conflict with that of an
5616 # existing column. note that this implies that any Column
5617 # must **not** set up its _label before its parent table has
5618 # all of its other Column objects set up. There are several
5619 # tables in the test suite which will fail otherwise; example:
5620 # table "owner" has columns "name" and "owner_name". Therefore
5621 # column owner.name cannot use the label "owner_name", it has
5622 # to be "owner_name_1".
5623 if label in t.c:
5624 _label = label
5625 counter = 1
5626 while _label in t.c:
5627 _label = label + f"_{counter}"
5628 counter += 1
5629 label = _label
5631 return coercions.expect(roles.TruncatedLabelRole, label)
5633 else:
5634 return name
5636 def _make_proxy(
5637 self,
5638 selectable: FromClause,
5639 *,
5640 primary_key: ColumnSet,
5641 foreign_keys: Set[KeyedColumnElement[Any]],
5642 name: Optional[str] = None,
5643 key: Optional[str] = None,
5644 name_is_truncatable: bool = False,
5645 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5646 disallow_is_literal: bool = False,
5647 **kw: Any,
5648 ) -> typing_Tuple[str, ColumnClause[_T]]:
5649 # the "is_literal" flag normally should never be propagated; a proxied
5650 # column is always a SQL identifier and never the actual expression
5651 # being evaluated. however, there is a case where the "is_literal" flag
5652 # might be used to allow the given identifier to have a fixed quoting
5653 # pattern already, so maintain the flag for the proxy unless a
5654 # :class:`.Label` object is creating the proxy. See [ticket:4730].
5655 is_literal = (
5656 not disallow_is_literal
5657 and self.is_literal
5658 and (
5659 # note this does not accommodate for quoted_name differences
5660 # right now
5661 name is None
5662 or name == self.name
5663 )
5664 )
5665 c = self._constructor(
5666 (
5667 coercions.expect(roles.TruncatedLabelRole, name or self.name)
5668 if name_is_truncatable
5669 else (name or self.name)
5670 ),
5671 type_=self.type,
5672 _selectable=selectable,
5673 is_literal=is_literal,
5674 )
5675 c._propagate_attrs = selectable._propagate_attrs
5676 if name is None:
5677 c.key = self.key
5678 if compound_select_cols:
5679 c._proxies = list(compound_select_cols)
5680 else:
5681 c._proxies = [self]
5683 if selectable._is_clone_of is not None:
5684 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5685 return c.key, c
5688class TableValuedColumn(NamedColumn[_T]):
5689 __visit_name__ = "table_valued_column"
5691 _traverse_internals: _TraverseInternalsType = [
5692 ("name", InternalTraversal.dp_anon_name),
5693 ("type", InternalTraversal.dp_type),
5694 ("scalar_alias", InternalTraversal.dp_clauseelement),
5695 ]
5697 def __init__(self, scalar_alias: NamedFromClause, type_: TypeEngine[_T]):
5698 self.scalar_alias = scalar_alias
5699 self.key = self.name = scalar_alias.name
5700 self.type = type_
5702 def _copy_internals(
5703 self, clone: _CloneCallableType = _clone, **kw: Any
5704 ) -> None:
5705 self.scalar_alias = clone(self.scalar_alias, **kw)
5706 self.key = self.name = self.scalar_alias.name
5708 @util.ro_non_memoized_property
5709 def _from_objects(self) -> List[FromClause]:
5710 return [self.scalar_alias]
5713class CollationClause(ColumnElement[str]):
5714 __visit_name__ = "collation"
5716 _traverse_internals: _TraverseInternalsType = [
5717 ("collation", InternalTraversal.dp_string),
5718 ("collation_schema", InternalTraversal.dp_string),
5719 ]
5721 @classmethod
5722 @util.preload_module("sqlalchemy.sql.sqltypes")
5723 def _create_collation_expression(
5724 cls,
5725 expression: _ColumnExpressionArgument[str],
5726 collation: str,
5727 collation_schema: Optional[str] = None,
5728 ) -> BinaryExpression[str]:
5730 sqltypes = util.preloaded.sql_sqltypes
5732 expr = coercions.expect(roles.ExpressionElementRole[str], expression)
5734 if expr.type._type_affinity is sqltypes.String:
5735 collate_type = expr.type._with_collation(
5736 collation, collation_schema
5737 )
5738 else:
5739 collate_type = expr.type
5741 return BinaryExpression(
5742 expr,
5743 CollationClause(collation, collation_schema),
5744 operators.collate,
5745 type_=collate_type,
5746 )
5748 def __init__(self, collation, collation_schema=None):
5749 self.collation = collation
5750 self.collation_schema = collation_schema
5753class _IdentifiedClause(Executable, ClauseElement):
5754 __visit_name__ = "identified"
5756 def __init__(self, ident):
5757 self.ident = ident
5760class SavepointClause(_IdentifiedClause):
5761 __visit_name__ = "savepoint"
5762 inherit_cache = False
5765class RollbackToSavepointClause(_IdentifiedClause):
5766 __visit_name__ = "rollback_to_savepoint"
5767 inherit_cache = False
5770class ReleaseSavepointClause(_IdentifiedClause):
5771 __visit_name__ = "release_savepoint"
5772 inherit_cache = False
5775class quoted_name(util.MemoizedSlots, str):
5776 """Represent a SQL identifier combined with quoting preferences.
5778 :class:`.quoted_name` is a Python unicode/str subclass which
5779 represents a particular identifier name along with a
5780 ``quote`` flag. This ``quote`` flag, when set to
5781 ``True`` or ``False``, overrides automatic quoting behavior
5782 for this identifier in order to either unconditionally quote
5783 or to not quote the name. If left at its default of ``None``,
5784 quoting behavior is applied to the identifier on a per-backend basis
5785 based on an examination of the token itself.
5787 A :class:`.quoted_name` object with ``quote=True`` is also
5788 prevented from being modified in the case of a so-called
5789 "name normalize" option. Certain database backends, such as
5790 Oracle Database, Firebird, and DB2 "normalize" case-insensitive names
5791 as uppercase. The SQLAlchemy dialects for these backends
5792 convert from SQLAlchemy's lower-case-means-insensitive convention
5793 to the upper-case-means-insensitive conventions of those backends.
5794 The ``quote=True`` flag here will prevent this conversion from occurring
5795 to support an identifier that's quoted as all lower case against
5796 such a backend.
5798 The :class:`.quoted_name` object is normally created automatically
5799 when specifying the name for key schema constructs such as
5800 :class:`_schema.Table`, :class:`_schema.Column`, and others.
5801 The class can also be
5802 passed explicitly as the name to any function that receives a name which
5803 can be quoted, such as :meth:`.Inspector.has_table` with an
5804 unconditionally quoted name::
5806 from sqlalchemy import create_engine
5807 from sqlalchemy import inspect
5808 from sqlalchemy.sql import quoted_name
5810 engine = create_engine("oracle+oracledb://some_dsn")
5811 print(inspect(engine).has_table(quoted_name("some_table", True)))
5813 The above logic will run the "has table" logic against the Oracle Database
5814 backend, passing the name exactly as ``"some_table"`` without converting to
5815 upper case.
5817 A :class:`.quoted_name` object with ``quote=False`` may be passed to APIs
5818 that apply automatic quoting in order to keep the given name unquoted,
5819 such as when a PostgreSQL ``INHERITS`` option refers to a schema-qualified
5820 table name like ``my_schema.some_table``.
5822 """
5824 __slots__ = "quote", "lower", "upper"
5826 quote: Optional[bool]
5828 @overload
5829 @classmethod
5830 def construct(cls, value: str, quote: Optional[bool]) -> quoted_name: ...
5832 @overload
5833 @classmethod
5834 def construct(cls, value: None, quote: Optional[bool]) -> None: ...
5836 @classmethod
5837 def construct(
5838 cls, value: Optional[str], quote: Optional[bool]
5839 ) -> Optional[quoted_name]:
5840 if value is None:
5841 return None
5842 else:
5843 return quoted_name(value, quote)
5845 def __new__(cls, value: str, quote: Optional[bool]) -> Self:
5846 assert (
5847 value is not None
5848 ), "use quoted_name.construct() for None passthrough"
5849 if isinstance(value, cls) and (quote is None or value.quote == quote):
5850 return value
5851 self = super().__new__(cls, value)
5853 self.quote = quote
5854 return self
5856 def __reduce__(self):
5857 return quoted_name, (str(self), self.quote)
5859 def _memoized_method_lower(self):
5860 if self.quote:
5861 return self
5862 else:
5863 return str(self).lower()
5865 def _memoized_method_upper(self):
5866 if self.quote:
5867 return self
5868 else:
5869 return str(self).upper()
5872def _find_columns(clause: ClauseElement) -> Set[ColumnClause[Any]]:
5873 """locate Column objects within the given expression."""
5875 cols: Set[ColumnClause[Any]] = set()
5876 traverse(clause, {}, {"column": cols.add})
5877 return cols
5880def _type_from_args(args: Sequence[ColumnElement[_T]]) -> TypeEngine[_T]:
5881 for a in args:
5882 if not a.type._isnull:
5883 return a.type
5884 else:
5885 return type_api.NULLTYPE # type: ignore[return-value]
5888def _corresponding_column_or_error(fromclause, column, require_embedded=False):
5889 c = fromclause.corresponding_column(
5890 column, require_embedded=require_embedded
5891 )
5892 if c is None:
5893 raise exc.InvalidRequestError(
5894 "Given column '%s', attached to table '%s', "
5895 "failed to locate a corresponding column from table '%s'"
5896 % (column, getattr(column, "table", None), fromclause.description)
5897 )
5898 return c
5901class _memoized_property_but_not_nulltype(
5902 util.memoized_property["TypeEngine[_T]"]
5903):
5904 """memoized property, but dont memoize NullType"""
5906 def __get__(self, obj, cls):
5907 if obj is None:
5908 return self
5909 result = self.fget(obj)
5910 if not result._isnull:
5911 obj.__dict__[self.__name__] = result
5912 return result
5915class AnnotatedColumnElement(Annotated):
5916 _Annotated__element: ColumnElement[Any]
5918 def __init__(self, element, values):
5919 Annotated.__init__(self, element, values)
5920 for attr in (
5921 "comparator",
5922 "_proxy_key",
5923 "_tq_key_label",
5924 "_tq_label",
5925 "_non_anon_label",
5926 "type",
5927 ):
5928 self.__dict__.pop(attr, None)
5929 for attr in ("name", "key", "table"):
5930 if self.__dict__.get(attr, False) is None:
5931 self.__dict__.pop(attr)
5933 def _with_annotations(self, values):
5934 clone = super()._with_annotations(values)
5935 for attr in (
5936 "comparator",
5937 "_proxy_key",
5938 "_tq_key_label",
5939 "_tq_label",
5940 "_non_anon_label",
5941 ):
5942 clone.__dict__.pop(attr, None)
5943 return clone
5945 @util.memoized_property
5946 def name(self):
5947 """pull 'name' from parent, if not present"""
5948 return self._Annotated__element.name
5950 @_memoized_property_but_not_nulltype
5951 def type(self):
5952 """pull 'type' from parent and don't cache if null.
5954 type is routinely changed on existing columns within the
5955 mapped_column() initialization process, and "type" is also consulted
5956 during the creation of SQL expressions. Therefore it can change after
5957 it was already retrieved. At the same time we don't want annotated
5958 objects having overhead when expressions are produced, so continue
5959 to memoize, but only when we have a non-null type.
5961 """
5962 return self._Annotated__element.type
5964 @util.memoized_property
5965 def table(self):
5966 """pull 'table' from parent, if not present"""
5967 return self._Annotated__element.table
5969 @util.memoized_property
5970 def key(self):
5971 """pull 'key' from parent, if not present"""
5972 return self._Annotated__element.key
5974 @util.memoized_property
5975 def info(self) -> _InfoType:
5976 if TYPE_CHECKING:
5977 assert isinstance(self._Annotated__element, Column)
5978 return self._Annotated__element.info
5980 @util.memoized_property
5981 def _anon_name_label(self) -> str:
5982 return self._Annotated__element._anon_name_label
5985class _truncated_label(quoted_name):
5986 """A unicode subclass used to identify symbolic "
5987 "names that may require truncation."""
5989 __slots__ = ()
5991 def __new__(cls, value: str, quote: Optional[bool] = None) -> Self:
5992 quote = getattr(value, "quote", quote)
5993 # return super(_truncated_label, cls).__new__(cls, value, quote, True)
5994 return super().__new__(cls, value, quote)
5996 def __reduce__(self) -> Any:
5997 return self.__class__, (str(self), self.quote)
5999 def apply_map(self, map_: Mapping[str, Any]) -> str:
6000 return self
6003class conv(_truncated_label):
6004 """Mark a string indicating that a name has already been converted
6005 by a naming convention.
6007 This is a string subclass that indicates a name that should not be
6008 subject to any further naming conventions.
6010 E.g. when we create a :class:`.Constraint` using a naming convention
6011 as follows::
6013 m = MetaData(
6014 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
6015 )
6016 t = Table(
6017 "t", m, Column("x", Integer), CheckConstraint("x > 5", name="x5")
6018 )
6020 The name of the above constraint will be rendered as ``"ck_t_x5"``.
6021 That is, the existing name ``x5`` is used in the naming convention as the
6022 ``constraint_name`` token.
6024 In some situations, such as in migration scripts, we may be rendering
6025 the above :class:`.CheckConstraint` with a name that's already been
6026 converted. In order to make sure the name isn't double-modified, the
6027 new name is applied using the :func:`_schema.conv` marker. We can
6028 use this explicitly as follows::
6031 m = MetaData(
6032 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
6033 )
6034 t = Table(
6035 "t",
6036 m,
6037 Column("x", Integer),
6038 CheckConstraint("x > 5", name=conv("ck_t_x5")),
6039 )
6041 Where above, the :func:`_schema.conv` marker indicates that the constraint
6042 name here is final, and the name will render as ``"ck_t_x5"`` and not
6043 ``"ck_t_ck_t_x5"``
6045 .. seealso::
6047 :ref:`constraint_naming_conventions`
6049 """
6051 __slots__ = ()
6054# for backwards compatibility in case
6055# someone is re-implementing the
6056# _truncated_identifier() sequence in a custom
6057# compiler
6058_generated_label = _truncated_label
6059_anonymous_label_escape = re.compile(r"[%\(\) \$]+")
6060# for bind parameter keys, additionally escape the characters that
6061# SQLCompiler.bindname_escape_characters would otherwise escape only at
6062# compile time, after the uniquifying counter has already been applied.
6063# escaping them up front is what allows names like "a.b" and "a_b" to be
6064# disambiguated as "a_b_1" / "a_b_2" rather than colliding. see #13534
6065_bind_key_escape = re.compile(r"[%\(\) \$\.\[\]:]+")
6068class _anonymous_label(_truncated_label):
6069 """A unicode subclass used to identify anonymously
6070 generated names."""
6072 __slots__ = ()
6074 @classmethod
6075 def safe_construct_with_key(
6076 cls, seed: int | str, body: str, sanitize_key: bool = False
6077 ) -> typing_Tuple[_anonymous_label, str]:
6078 # need to escape chars that interfere with format
6079 # strings in any case, issue #8724
6080 if sanitize_key:
6081 # sanitize_key is an extra step used by BindParameter, which
6082 # also escapes the characters that would otherwise be escaped
6083 # only at compile time; issue #13534
6084 body = _bind_key_escape.sub("_", body).strip("_")
6085 else:
6086 body = _anonymous_label_escape.sub("_", body)
6088 key = f"{seed} {body.replace('%', '%%')}"
6089 label = _anonymous_label(f"%({key})s")
6090 return label, key
6092 @classmethod
6093 def safe_construct(
6094 cls, seed: int | str, body: str, sanitize_key: bool = False
6095 ) -> _anonymous_label:
6096 # need to escape chars that interfere with format
6097 # strings in any case, issue #8724
6098 if sanitize_key:
6099 # sanitize_key is an extra step used by BindParameter, which
6100 # also escapes the characters that would otherwise be escaped
6101 # only at compile time; issue #13534
6102 body = _bind_key_escape.sub("_", body).strip("_")
6103 else:
6104 body = _anonymous_label_escape.sub("_", body)
6106 return _anonymous_label(f"%({seed} {body.replace('%', '%%')})s")
6108 def __add__(self, other: str) -> _anonymous_label:
6109 if "%" in other and not isinstance(other, _anonymous_label):
6110 other = str(other).replace("%", "%%")
6111 else:
6112 other = str(other)
6114 return _anonymous_label(
6115 quoted_name(
6116 str.__add__(self, other),
6117 self.quote,
6118 )
6119 )
6121 def __radd__(self, other: str) -> _anonymous_label:
6122 if "%" in other and not isinstance(other, _anonymous_label):
6123 other = str(other).replace("%", "%%")
6124 else:
6125 other = str(other)
6127 return _anonymous_label(
6128 quoted_name(
6129 str.__add__(other, self),
6130 self.quote,
6131 )
6132 )
6134 def apply_map(self, map_: Mapping[str, Any]) -> str:
6135 if self.quote is not None:
6136 # preserve quoting only if necessary
6137 return quoted_name(self % map_, self.quote)
6138 else:
6139 # else skip the constructor call
6140 return self % map_