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-2025 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 Generative
62from .base import HasMemoized
63from .base import Immutable
64from .base import NO_ARG
65from .base import SingletonConstant
66from .cache_key import MemoizedHasCacheKey
67from .cache_key import NO_CACHE
68from .coercions import _document_text_coercion # noqa
69from .operators import ColumnOperators
70from .operators import OperatorClass
71from .traversals import HasCopyInternals
72from .visitors import cloned_traverse
73from .visitors import ExternallyTraversible
74from .visitors import InternalTraversal
75from .visitors import traverse
76from .visitors import Visitable
77from .. import exc
78from .. import inspection
79from .. import util
80from ..util import HasMemoized_ro_memoized_attribute
81from ..util import TypingOnly
82from ..util.typing import Self
83from ..util.typing import TupleAny
84from ..util.typing import Unpack
87if typing.TYPE_CHECKING:
88 from ._typing import _ByArgument
89 from ._typing import _ColumnExpressionArgument
90 from ._typing import _ColumnExpressionOrStrLabelArgument
91 from ._typing import _DMLOnlyColumnArgument
92 from ._typing import _HasDialect
93 from ._typing import _InfoType
94 from ._typing import _PropagateAttrsType
95 from ._typing import _TypeEngineArgument
96 from .base import _EntityNamespace
97 from .base import ColumnSet
98 from .cache_key import _CacheKeyTraversalType
99 from .cache_key import CacheKey
100 from .compiler import Compiled
101 from .compiler import SQLCompiler
102 from .functions import FunctionElement
103 from .operators import OperatorType
104 from .schema import Column
105 from .schema import DefaultGenerator
106 from .schema import FetchedValue
107 from .schema import ForeignKey
108 from .selectable import _SelectIterable
109 from .selectable import FromClause
110 from .selectable import NamedFromClause
111 from .selectable import TextualSelect
112 from .sqltypes import TupleType
113 from .type_api import TypeEngine
114 from .visitors import _CloneCallableType
115 from .visitors import _TraverseInternalsType
116 from .visitors import anon_map
117 from ..engine import Connection
118 from ..engine import Dialect
119 from ..engine.interfaces import _CoreMultiExecuteParams
120 from ..engine.interfaces import CacheStats
121 from ..engine.interfaces import CompiledCacheType
122 from ..engine.interfaces import CoreExecuteOptionsParameter
123 from ..engine.interfaces import SchemaTranslateMapType
124 from ..engine.result import Result
127_NUMERIC = Union[float, Decimal]
128_NUMBER = Union[float, int, Decimal]
130_T = TypeVar("_T", bound="Any")
131_T_co = TypeVar("_T_co", bound=Any, covariant=True)
132_OPT = TypeVar("_OPT", bound="Any")
133_NT = TypeVar("_NT", bound="_NUMERIC")
135_NMT = TypeVar("_NMT", bound="_NUMBER")
138@overload
139def literal(
140 value: Any,
141 type_: _TypeEngineArgument[_T],
142 literal_execute: bool = False,
143) -> BindParameter[_T]: ...
146@overload
147def literal(
148 value: _T,
149 type_: None = None,
150 literal_execute: bool = False,
151) -> BindParameter[_T]: ...
154@overload
155def literal(
156 value: Any,
157 type_: Optional[_TypeEngineArgument[Any]] = None,
158 literal_execute: bool = False,
159) -> BindParameter[Any]: ...
162def literal(
163 value: Any,
164 type_: Optional[_TypeEngineArgument[Any]] = None,
165 literal_execute: bool = False,
166) -> BindParameter[Any]:
167 r"""Return a literal clause, bound to a bind parameter.
169 Literal clauses are created automatically when non-
170 :class:`_expression.ClauseElement` objects (such as strings, ints, dates,
171 etc.) are
172 used in a comparison operation with a :class:`_expression.ColumnElement`
173 subclass,
174 such as a :class:`~sqlalchemy.schema.Column` object. Use this function
175 to force the generation of a literal clause, which will be created as a
176 :class:`BindParameter` with a bound value.
178 :param value: the value to be bound. Can be any Python object supported by
179 the underlying DB-API, or is translatable via the given type argument.
181 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine` which will
182 provide bind-parameter translation for this literal.
184 :param literal_execute: optional bool, when True, the SQL engine will
185 attempt to render the bound value directly in the SQL statement at
186 execution time rather than providing as a parameter value.
188 .. versionadded:: 2.0
190 """
191 return coercions.expect(
192 roles.LiteralValueRole,
193 value,
194 type_=type_,
195 literal_execute=literal_execute,
196 )
199def literal_column(
200 text: str, type_: Optional[_TypeEngineArgument[_T]] = None
201) -> ColumnClause[_T]:
202 r"""Produce a :class:`.ColumnClause` object that has the
203 :paramref:`_expression.column.is_literal` flag set to True.
205 :func:`_expression.literal_column` is similar to
206 :func:`_expression.column`, except that
207 it is more often used as a "standalone" column expression that renders
208 exactly as stated; while :func:`_expression.column`
209 stores a string name that
210 will be assumed to be part of a table and may be quoted as such,
211 :func:`_expression.literal_column` can be that,
212 or any other arbitrary column-oriented
213 expression.
215 :param text: the text of the expression; can be any SQL expression.
216 Quoting rules will not be applied. To specify a column-name expression
217 which should be subject to quoting rules, use the :func:`column`
218 function.
220 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine`
221 object which will
222 provide result-set translation and additional expression semantics for
223 this column. If left as ``None`` the type will be :class:`.NullType`.
225 .. seealso::
227 :func:`_expression.column`
229 :func:`_expression.text`
231 :ref:`tutorial_select_arbitrary_text`
233 """
234 return ColumnClause(text, type_=type_, is_literal=True)
237class CompilerElement(Visitable):
238 """base class for SQL elements that can be compiled to produce a
239 SQL string.
241 .. versionadded:: 2.0
243 """
245 __slots__ = ()
246 __visit_name__ = "compiler_element"
248 supports_execution = False
250 stringify_dialect = "default"
252 @util.preload_module("sqlalchemy.engine.default")
253 @util.preload_module("sqlalchemy.engine.url")
254 def compile(
255 self,
256 bind: Optional[_HasDialect] = None,
257 dialect: Optional[Dialect] = None,
258 **kw: Any,
259 ) -> Compiled:
260 """Compile this SQL expression.
262 The return value is a :class:`~.Compiled` object.
263 Calling ``str()`` or ``unicode()`` on the returned value will yield a
264 string representation of the result. The
265 :class:`~.Compiled` object also can return a
266 dictionary of bind parameter names and values
267 using the ``params`` accessor.
269 :param bind: An :class:`.Connection` or :class:`.Engine` which
270 can provide a :class:`.Dialect` in order to generate a
271 :class:`.Compiled` object. If the ``bind`` and
272 ``dialect`` parameters are both omitted, a default SQL compiler
273 is used.
275 :param column_keys: Used for INSERT and UPDATE statements, a list of
276 column names which should be present in the VALUES clause of the
277 compiled statement. If ``None``, all columns from the target table
278 object are rendered.
280 :param dialect: A :class:`.Dialect` instance which can generate
281 a :class:`.Compiled` object. This argument takes precedence over
282 the ``bind`` argument.
284 :param compile_kwargs: optional dictionary of additional parameters
285 that will be passed through to the compiler within all "visit"
286 methods. This allows any custom flag to be passed through to
287 a custom compilation construct, for example. It is also used
288 for the case of passing the ``literal_binds`` flag through::
290 from sqlalchemy.sql import table, column, select
292 t = table("t", column("x"))
294 s = select(t).where(t.c.x == 5)
296 print(s.compile(compile_kwargs={"literal_binds": True}))
298 .. seealso::
300 :ref:`faq_sql_expression_string`
302 """
304 if dialect is None:
305 if bind:
306 dialect = bind.dialect
307 elif self.stringify_dialect == "default":
308 dialect = self._default_dialect()
309 else:
310 url = util.preloaded.engine_url
311 dialect = url.URL.create(
312 self.stringify_dialect
313 ).get_dialect()()
315 return self._compiler(dialect, **kw)
317 def _default_dialect(self):
318 default = util.preloaded.engine_default
319 return default.StrCompileDialect()
321 def _compiler(self, dialect: Dialect, **kw: Any) -> Compiled:
322 """Return a compiler appropriate for this ClauseElement, given a
323 Dialect."""
325 if TYPE_CHECKING:
326 assert isinstance(self, ClauseElement)
327 return dialect.statement_compiler(dialect, self, **kw)
329 def __str__(self) -> str:
330 return str(self.compile())
333@inspection._self_inspects
334class ClauseElement(
335 SupportsWrappingAnnotations,
336 MemoizedHasCacheKey,
337 HasCopyInternals,
338 ExternallyTraversible,
339 CompilerElement,
340):
341 """Base class for elements of a programmatically constructed SQL
342 expression.
344 """
346 __visit_name__ = "clause"
348 if TYPE_CHECKING:
350 @util.memoized_property
351 def _propagate_attrs(self) -> _PropagateAttrsType:
352 """like annotations, however these propagate outwards liberally
353 as SQL constructs are built, and are set up at construction time.
355 """
356 ...
358 else:
359 _propagate_attrs = util.EMPTY_DICT
361 @util.ro_memoized_property
362 def description(self) -> Optional[str]:
363 return None
365 _is_clone_of: Optional[Self] = None
367 is_clause_element = True
368 is_selectable = False
369 is_dml = False
370 _is_column_element = False
371 _is_keyed_column_element = False
372 _is_table = False
373 _gen_static_annotations_cache_key = False
374 _is_textual = False
375 _is_from_clause = False
376 _is_returns_rows = False
377 _is_text_clause = False
378 _is_from_container = False
379 _is_select_container = False
380 _is_select_base = False
381 _is_select_statement = False
382 _is_bind_parameter = False
383 _is_clause_list = False
384 _is_lambda_element = False
385 _is_singleton_constant = False
386 _is_immutable = False
387 _is_star = False
389 @property
390 def _order_by_label_element(self) -> Optional[Label[Any]]:
391 return None
393 _cache_key_traversal: _CacheKeyTraversalType = None
395 negation_clause: ColumnElement[bool]
397 if typing.TYPE_CHECKING:
399 def get_children(
400 self, *, omit_attrs: typing_Tuple[str, ...] = ..., **kw: Any
401 ) -> Iterable[ClauseElement]: ...
403 @util.ro_non_memoized_property
404 def _from_objects(self) -> List[FromClause]:
405 return []
407 def _set_propagate_attrs(self, values: Mapping[str, Any]) -> Self:
408 # usually, self._propagate_attrs is empty here. one case where it's
409 # not is a subquery against ORM select, that is then pulled as a
410 # property of an aliased class. should all be good
412 # assert not self._propagate_attrs
414 self._propagate_attrs = util.immutabledict(values)
415 return self
417 def _default_compiler(self) -> SQLCompiler:
418 dialect = self._default_dialect()
419 return dialect.statement_compiler(dialect, self) # type: ignore
421 def _clone(self, **kw: Any) -> Self:
422 """Create a shallow copy of this ClauseElement.
424 This method may be used by a generative API. Its also used as
425 part of the "deep" copy afforded by a traversal that combines
426 the _copy_internals() method.
428 """
430 skip = self._memoized_keys
431 c = self.__class__.__new__(self.__class__)
433 if skip:
434 # ensure this iteration remains atomic
435 c.__dict__ = {
436 k: v for k, v in self.__dict__.copy().items() if k not in skip
437 }
438 else:
439 c.__dict__ = self.__dict__.copy()
441 # this is a marker that helps to "equate" clauses to each other
442 # when a Select returns its list of FROM clauses. the cloning
443 # process leaves around a lot of remnants of the previous clause
444 # typically in the form of column expressions still attached to the
445 # old table.
446 cc = self._is_clone_of
447 c._is_clone_of = cc if cc is not None else self
448 return c
450 def _negate_in_binary(self, negated_op, original_op):
451 """a hook to allow the right side of a binary expression to respond
452 to a negation of the binary expression.
454 Used for the special case of expanding bind parameter with IN.
456 """
457 return self
459 def _with_binary_element_type(self, type_):
460 """in the context of binary expression, convert the type of this
461 object to the one given.
463 applies only to :class:`_expression.ColumnElement` classes.
465 """
466 return self
468 @property
469 def _constructor(self): # type: ignore[override]
470 """return the 'constructor' for this ClauseElement.
472 This is for the purposes for creating a new object of
473 this type. Usually, its just the element's __class__.
474 However, the "Annotated" version of the object overrides
475 to return the class of its proxied element.
477 """
478 return self.__class__
480 @HasMemoized.memoized_attribute
481 def _cloned_set(self):
482 """Return the set consisting all cloned ancestors of this
483 ClauseElement.
485 Includes this ClauseElement. This accessor tends to be used for
486 FromClause objects to identify 'equivalent' FROM clauses, regardless
487 of transformative operations.
489 """
490 s = util.column_set()
491 f: Optional[ClauseElement] = self
493 # note this creates a cycle, asserted in test_memusage. however,
494 # turning this into a plain @property adds tends of thousands of method
495 # calls to Core / ORM performance tests, so the small overhead
496 # introduced by the relatively small amount of short term cycles
497 # produced here is preferable
498 while f is not None:
499 s.add(f)
500 f = f._is_clone_of
501 return s
503 def _de_clone(self):
504 while self._is_clone_of is not None:
505 self = self._is_clone_of
506 return self
508 @util.ro_non_memoized_property
509 def entity_namespace(self) -> _EntityNamespace:
510 raise AttributeError(
511 "This SQL expression has no entity namespace "
512 "with which to filter from."
513 )
515 def __getstate__(self):
516 d = self.__dict__.copy()
517 d.pop("_is_clone_of", None)
518 d.pop("_generate_cache_key", None)
519 return d
521 def _execute_on_connection(
522 self,
523 connection: Connection,
524 distilled_params: _CoreMultiExecuteParams,
525 execution_options: CoreExecuteOptionsParameter,
526 ) -> Result[Unpack[TupleAny]]:
527 if self.supports_execution:
528 if TYPE_CHECKING:
529 assert isinstance(self, Executable)
530 return connection._execute_clauseelement(
531 self, distilled_params, execution_options
532 )
533 else:
534 raise exc.ObjectNotExecutableError(self)
536 def _execute_on_scalar(
537 self,
538 connection: Connection,
539 distilled_params: _CoreMultiExecuteParams,
540 execution_options: CoreExecuteOptionsParameter,
541 ) -> Any:
542 """an additional hook for subclasses to provide a different
543 implementation for connection.scalar() vs. connection.execute().
545 .. versionadded:: 2.0
547 """
548 return self._execute_on_connection(
549 connection, distilled_params, execution_options
550 ).scalar()
552 def _get_embedded_bindparams(self) -> Sequence[BindParameter[Any]]:
553 """Return the list of :class:`.BindParameter` objects embedded in the
554 object.
556 This accomplishes the same purpose as ``visitors.traverse()`` or
557 similar would provide, however by making use of the cache key
558 it takes advantage of memoization of the key to result in fewer
559 net method calls, assuming the statement is also going to be
560 executed.
562 """
564 key = self._generate_cache_key()
565 if key is None:
566 bindparams: List[BindParameter[Any]] = []
568 traverse(self, {}, {"bindparam": bindparams.append})
569 return bindparams
571 else:
572 return key.bindparams
574 def unique_params(
575 self,
576 __optionaldict: Optional[Dict[str, Any]] = None,
577 /,
578 **kwargs: Any,
579 ) -> Self:
580 """Return a copy with :func:`_expression.bindparam` elements
581 replaced.
583 Same functionality as :meth:`_expression.ClauseElement.params`,
584 except adds `unique=True`
585 to affected bind parameters so that multiple statements can be
586 used.
588 """
589 return self._replace_params(True, __optionaldict, kwargs)
591 def params(
592 self,
593 __optionaldict: Optional[Mapping[str, Any]] = None,
594 /,
595 **kwargs: Any,
596 ) -> Self:
597 """Return a copy with :func:`_expression.bindparam` elements
598 replaced.
600 Returns a copy of this ClauseElement with
601 :func:`_expression.bindparam`
602 elements replaced with values taken from the given dictionary::
604 >>> clause = column("x") + bindparam("foo")
605 >>> print(clause.compile().params)
606 {'foo':None}
607 >>> print(clause.params({"foo": 7}).compile().params)
608 {'foo':7}
610 """
611 return self._replace_params(False, __optionaldict, kwargs)
613 def _replace_params(
614 self,
615 unique: bool,
616 optionaldict: Optional[Mapping[str, Any]],
617 kwargs: Dict[str, Any],
618 ) -> Self:
619 if optionaldict:
620 kwargs.update(optionaldict)
622 def visit_bindparam(bind: BindParameter[Any]) -> None:
623 if bind.key in kwargs:
624 bind.value = kwargs[bind.key]
625 bind.required = False
626 if unique:
627 bind._convert_to_unique()
629 return cloned_traverse(
630 self,
631 {"maintain_key": True, "detect_subquery_cols": True},
632 {"bindparam": visit_bindparam},
633 )
635 def compare(self, other: ClauseElement, **kw: Any) -> bool:
636 r"""Compare this :class:`_expression.ClauseElement` to
637 the given :class:`_expression.ClauseElement`.
639 Subclasses should override the default behavior, which is a
640 straight identity comparison.
642 \**kw are arguments consumed by subclass ``compare()`` methods and
643 may be used to modify the criteria for comparison
644 (see :class:`_expression.ColumnElement`).
646 """
647 return traversals.compare(self, other, **kw)
649 def self_group(
650 self, against: Optional[OperatorType] = None
651 ) -> ClauseElement:
652 """Apply a 'grouping' to this :class:`_expression.ClauseElement`.
654 This method is overridden by subclasses to return a "grouping"
655 construct, i.e. parenthesis. In particular it's used by "binary"
656 expressions to provide a grouping around themselves when placed into a
657 larger expression, as well as by :func:`_expression.select`
658 constructs when placed into the FROM clause of another
659 :func:`_expression.select`. (Note that subqueries should be
660 normally created using the :meth:`_expression.Select.alias` method,
661 as many
662 platforms require nested SELECT statements to be named).
664 As expressions are composed together, the application of
665 :meth:`self_group` is automatic - end-user code should never
666 need to use this method directly. Note that SQLAlchemy's
667 clause constructs take operator precedence into account -
668 so parenthesis might not be needed, for example, in
669 an expression like ``x OR (y AND z)`` - AND takes precedence
670 over OR.
672 The base :meth:`self_group` method of
673 :class:`_expression.ClauseElement`
674 just returns self.
675 """
676 return self
678 def _ungroup(self) -> ClauseElement:
679 """Return this :class:`_expression.ClauseElement`
680 without any groupings.
681 """
683 return self
685 def _compile_w_cache(
686 self,
687 dialect: Dialect,
688 *,
689 compiled_cache: Optional[CompiledCacheType],
690 column_keys: List[str],
691 for_executemany: bool = False,
692 schema_translate_map: Optional[SchemaTranslateMapType] = None,
693 **kw: Any,
694 ) -> typing_Tuple[
695 Compiled, Optional[Sequence[BindParameter[Any]]], CacheStats
696 ]:
697 elem_cache_key: Optional[CacheKey]
699 if compiled_cache is not None and dialect._supports_statement_cache:
700 elem_cache_key = self._generate_cache_key()
701 else:
702 elem_cache_key = None
704 extracted_params: Optional[Sequence[BindParameter[Any]]]
705 if elem_cache_key is not None:
706 if TYPE_CHECKING:
707 assert compiled_cache is not None
709 cache_key, extracted_params = elem_cache_key
710 key = (
711 dialect,
712 cache_key,
713 tuple(column_keys),
714 bool(schema_translate_map),
715 for_executemany,
716 )
717 compiled_sql = compiled_cache.get(key)
719 if compiled_sql is None:
720 cache_hit = dialect.CACHE_MISS
721 compiled_sql = self._compiler(
722 dialect,
723 cache_key=elem_cache_key,
724 column_keys=column_keys,
725 for_executemany=for_executemany,
726 schema_translate_map=schema_translate_map,
727 **kw,
728 )
729 compiled_cache[key] = compiled_sql
730 else:
731 cache_hit = dialect.CACHE_HIT
732 else:
733 extracted_params = None
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 )
743 if not dialect._supports_statement_cache:
744 cache_hit = dialect.NO_DIALECT_SUPPORT
745 elif compiled_cache is None:
746 cache_hit = dialect.CACHING_DISABLED
747 else:
748 cache_hit = dialect.NO_CACHE_KEY
750 return compiled_sql, extracted_params, cache_hit
752 def __invert__(self):
753 # undocumented element currently used by the ORM for
754 # relationship.contains()
755 if hasattr(self, "negation_clause"):
756 return self.negation_clause
757 else:
758 return self._negate()
760 def _negate(self) -> ClauseElement:
761 # TODO: this code is uncovered and in all likelihood is not included
762 # in any codepath. So this should raise NotImplementedError in 2.1
763 grouped = self.self_group(against=operators.inv)
764 assert isinstance(grouped, ColumnElement)
765 return UnaryExpression(grouped, operator=operators.inv)
767 def __bool__(self):
768 raise TypeError("Boolean value of this clause is not defined")
770 def __repr__(self):
771 friendly = self.description
772 if friendly is None:
773 return object.__repr__(self)
774 else:
775 return "<%s.%s at 0x%x; %s>" % (
776 self.__module__,
777 self.__class__.__name__,
778 id(self),
779 friendly,
780 )
783class DQLDMLClauseElement(ClauseElement):
784 """represents a :class:`.ClauseElement` that compiles to a DQL or DML
785 expression, not DDL.
787 .. versionadded:: 2.0
789 """
791 if typing.TYPE_CHECKING:
793 def _compiler(self, dialect: Dialect, **kw: Any) -> SQLCompiler:
794 """Return a compiler appropriate for this ClauseElement, given a
795 Dialect."""
796 ...
798 def compile( # noqa: A001
799 self,
800 bind: Optional[_HasDialect] = None,
801 dialect: Optional[Dialect] = None,
802 **kw: Any,
803 ) -> SQLCompiler: ...
806class CompilerColumnElement(
807 roles.DMLColumnRole,
808 roles.DDLConstraintColumnRole,
809 roles.ColumnsClauseRole,
810 CompilerElement,
811):
812 """A compiler-only column element used for ad-hoc string compilations.
814 .. versionadded:: 2.0
816 """
818 __slots__ = ()
820 _propagate_attrs = util.EMPTY_DICT
821 _is_collection_aggregate = False
822 _is_implicitly_boolean = False
824 def _with_binary_element_type(self, type_):
825 raise NotImplementedError()
827 def _gen_cache_key(self, anon_map, bindparams):
828 raise NotImplementedError()
830 @property
831 def _from_objects(self) -> List[FromClause]:
832 raise NotImplementedError()
835# SQLCoreOperations should be suiting the ExpressionElementRole
836# and ColumnsClauseRole. however the MRO issues become too elaborate
837# at the moment.
838class SQLCoreOperations(Generic[_T_co], ColumnOperators, TypingOnly):
839 __slots__ = ()
841 # annotations for comparison methods
842 # these are from operators->Operators / ColumnOperators,
843 # redefined with the specific types returned by ColumnElement hierarchies
844 if typing.TYPE_CHECKING:
846 @util.non_memoized_property
847 def _propagate_attrs(self) -> _PropagateAttrsType: ...
849 def operate(
850 self, op: OperatorType, *other: Any, **kwargs: Any
851 ) -> ColumnElement[Any]: ...
853 def reverse_operate(
854 self, op: OperatorType, other: Any, **kwargs: Any
855 ) -> ColumnElement[Any]: ...
857 @overload
858 def op(
859 self,
860 opstring: str,
861 precedence: int = ...,
862 is_comparison: bool = ...,
863 *,
864 return_type: _TypeEngineArgument[_OPT],
865 python_impl: Optional[Callable[..., Any]] = None,
866 operator_class: OperatorClass = ...,
867 ) -> Callable[[Any], BinaryExpression[_OPT]]: ...
869 @overload
870 def op(
871 self,
872 opstring: str,
873 precedence: int = ...,
874 is_comparison: bool = ...,
875 return_type: Optional[_TypeEngineArgument[Any]] = ...,
876 python_impl: Optional[Callable[..., Any]] = ...,
877 operator_class: OperatorClass = ...,
878 ) -> Callable[[Any], BinaryExpression[Any]]: ...
880 def op(
881 self,
882 opstring: str,
883 precedence: int = 0,
884 is_comparison: bool = False,
885 return_type: Optional[_TypeEngineArgument[Any]] = None,
886 python_impl: Optional[Callable[..., Any]] = None,
887 operator_class: OperatorClass = OperatorClass.BASE,
888 ) -> Callable[[Any], BinaryExpression[Any]]: ...
890 def bool_op(
891 self,
892 opstring: str,
893 precedence: int = 0,
894 python_impl: Optional[Callable[..., Any]] = None,
895 ) -> Callable[[Any], BinaryExpression[bool]]: ...
897 def __and__(self, other: Any) -> BooleanClauseList: ...
899 def __or__(self, other: Any) -> BooleanClauseList: ...
901 def __invert__(self) -> ColumnElement[_T_co]: ...
903 def __lt__(self, other: Any) -> ColumnElement[bool]: ...
905 def __le__(self, other: Any) -> ColumnElement[bool]: ...
907 # declare also that this class has an hash method otherwise
908 # it may be assumed to be None by type checkers since the
909 # object defines __eq__ and python sets it to None in that case:
910 # https://docs.python.org/3/reference/datamodel.html#object.__hash__
911 def __hash__(self) -> int: ...
913 def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
914 ...
916 def __ne__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
917 ...
919 def is_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
921 def is_not_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
923 def __gt__(self, other: Any) -> ColumnElement[bool]: ...
925 def __ge__(self, other: Any) -> ColumnElement[bool]: ...
927 def __neg__(self) -> UnaryExpression[_T_co]: ...
929 def __contains__(self, other: Any) -> ColumnElement[bool]: ...
931 def __getitem__(self, index: Any) -> ColumnElement[Any]: ...
933 @overload
934 def __lshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
936 @overload
937 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
939 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
941 @overload
942 def __rlshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
944 @overload
945 def __rlshift__(self, other: Any) -> ColumnElement[Any]: ...
947 def __rlshift__(self, other: Any) -> ColumnElement[Any]: ...
949 @overload
950 def __rshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
952 @overload
953 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
955 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
957 @overload
958 def __rrshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
960 @overload
961 def __rrshift__(self, other: Any) -> ColumnElement[Any]: ...
963 def __rrshift__(self, other: Any) -> ColumnElement[Any]: ...
965 def __matmul__(self, other: Any) -> ColumnElement[Any]: ...
967 def __rmatmul__(self, other: Any) -> ColumnElement[Any]: ...
969 @overload
970 def concat(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
972 @overload
973 def concat(self, other: Any) -> ColumnElement[Any]: ...
975 def concat(self, other: Any) -> ColumnElement[Any]: ...
977 def like(
978 self, other: Any, escape: Optional[str] = None
979 ) -> BinaryExpression[bool]: ...
981 def ilike(
982 self, other: Any, escape: Optional[str] = None
983 ) -> BinaryExpression[bool]: ...
985 def bitwise_xor(self, other: Any) -> BinaryExpression[Any]: ...
987 def bitwise_or(self, other: Any) -> BinaryExpression[Any]: ...
989 def bitwise_and(self, other: Any) -> BinaryExpression[Any]: ...
991 def bitwise_not(self) -> UnaryExpression[_T_co]: ...
993 def bitwise_lshift(self, other: Any) -> BinaryExpression[Any]: ...
995 def bitwise_rshift(self, other: Any) -> BinaryExpression[Any]: ...
997 def in_(
998 self,
999 other: Union[
1000 Iterable[Any], BindParameter[Any], roles.InElementRole
1001 ],
1002 ) -> BinaryExpression[bool]: ...
1004 def not_in(
1005 self,
1006 other: Union[
1007 Iterable[Any], BindParameter[Any], roles.InElementRole
1008 ],
1009 ) -> BinaryExpression[bool]: ...
1011 def notin_(
1012 self,
1013 other: Union[
1014 Iterable[Any], BindParameter[Any], roles.InElementRole
1015 ],
1016 ) -> BinaryExpression[bool]: ...
1018 def not_like(
1019 self, other: Any, escape: Optional[str] = None
1020 ) -> BinaryExpression[bool]: ...
1022 def notlike(
1023 self, other: Any, escape: Optional[str] = None
1024 ) -> BinaryExpression[bool]: ...
1026 def not_ilike(
1027 self, other: Any, escape: Optional[str] = None
1028 ) -> BinaryExpression[bool]: ...
1030 def notilike(
1031 self, other: Any, escape: Optional[str] = None
1032 ) -> BinaryExpression[bool]: ...
1034 def is_(self, other: Any) -> BinaryExpression[bool]: ...
1036 def is_not(self, other: Any) -> BinaryExpression[bool]: ...
1038 def isnot(self, other: Any) -> BinaryExpression[bool]: ...
1040 def startswith(
1041 self,
1042 other: Any,
1043 escape: Optional[str] = None,
1044 autoescape: bool = False,
1045 ) -> ColumnElement[bool]: ...
1047 def istartswith(
1048 self,
1049 other: Any,
1050 escape: Optional[str] = None,
1051 autoescape: bool = False,
1052 ) -> ColumnElement[bool]: ...
1054 def endswith(
1055 self,
1056 other: Any,
1057 escape: Optional[str] = None,
1058 autoescape: bool = False,
1059 ) -> ColumnElement[bool]: ...
1061 def iendswith(
1062 self,
1063 other: Any,
1064 escape: Optional[str] = None,
1065 autoescape: bool = False,
1066 ) -> ColumnElement[bool]: ...
1068 def contains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1070 def icontains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1072 def match(self, other: Any, **kwargs: Any) -> ColumnElement[bool]: ...
1074 def regexp_match(
1075 self, pattern: Any, flags: Optional[str] = None
1076 ) -> ColumnElement[bool]: ...
1078 def regexp_replace(
1079 self, pattern: Any, replacement: Any, flags: Optional[str] = None
1080 ) -> ColumnElement[str]: ...
1082 def desc(self) -> UnaryExpression[_T_co]: ...
1084 def asc(self) -> UnaryExpression[_T_co]: ...
1086 def nulls_first(self) -> UnaryExpression[_T_co]: ...
1088 def nullsfirst(self) -> UnaryExpression[_T_co]: ...
1090 def nulls_last(self) -> UnaryExpression[_T_co]: ...
1092 def nullslast(self) -> UnaryExpression[_T_co]: ...
1094 def collate(self, collation: str) -> CollationClause: ...
1096 def between(
1097 self, cleft: Any, cright: Any, symmetric: bool = False
1098 ) -> BinaryExpression[bool]: ...
1100 def distinct(self: _SQO[_T_co]) -> UnaryExpression[_T_co]: ...
1102 def any_(self) -> CollectionAggregate[Any]: ...
1104 def all_(self) -> CollectionAggregate[Any]: ...
1106 # numeric overloads. These need more tweaking
1107 # in particular they all need to have a variant for Optiona[_T]
1108 # because Optional only applies to the data side, not the expression
1109 # side
1111 @overload
1112 def __add__(
1113 self: _SQO[_NMT],
1114 other: Any,
1115 ) -> ColumnElement[_NMT]: ...
1117 @overload
1118 def __add__(
1119 self: _SQO[str],
1120 other: Any,
1121 ) -> ColumnElement[str]: ...
1123 @overload
1124 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1126 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1128 @overload
1129 def __radd__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1131 @overload
1132 def __radd__(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
1134 def __radd__(self, other: Any) -> ColumnElement[Any]: ...
1136 @overload
1137 def __sub__(
1138 self: _SQO[_NMT],
1139 other: Any,
1140 ) -> ColumnElement[_NMT]: ...
1142 @overload
1143 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1145 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1147 @overload
1148 def __rsub__(
1149 self: _SQO[_NMT],
1150 other: Any,
1151 ) -> ColumnElement[_NMT]: ...
1153 @overload
1154 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1156 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1158 @overload
1159 def __mul__(
1160 self: _SQO[_NMT],
1161 other: Any,
1162 ) -> ColumnElement[_NMT]: ...
1164 @overload
1165 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1167 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1169 @overload
1170 def __rmul__(
1171 self: _SQO[_NMT],
1172 other: Any,
1173 ) -> ColumnElement[_NMT]: ...
1175 @overload
1176 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1178 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1180 @overload
1181 def __mod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1183 @overload
1184 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1186 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1188 @overload
1189 def __rmod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1191 @overload
1192 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1194 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1196 @overload
1197 def __truediv__(
1198 self: _SQO[int], other: Any
1199 ) -> ColumnElement[_NUMERIC]: ...
1201 @overload
1202 def __truediv__(self: _SQO[_NT], other: Any) -> ColumnElement[_NT]: ...
1204 @overload
1205 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1207 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1209 @overload
1210 def __rtruediv__(
1211 self: _SQO[_NMT], other: Any
1212 ) -> ColumnElement[_NUMERIC]: ...
1214 @overload
1215 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1217 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1219 @overload
1220 def __floordiv__(
1221 self: _SQO[_NMT], other: Any
1222 ) -> ColumnElement[_NMT]: ...
1224 @overload
1225 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1227 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1229 @overload
1230 def __rfloordiv__(
1231 self: _SQO[_NMT], other: Any
1232 ) -> ColumnElement[_NMT]: ...
1234 @overload
1235 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1237 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1240class SQLColumnExpression(
1241 SQLCoreOperations[_T_co], roles.ExpressionElementRole[_T_co], TypingOnly
1242):
1243 """A type that may be used to indicate any SQL column element or object
1244 that acts in place of one.
1246 :class:`.SQLColumnExpression` is a base of
1247 :class:`.ColumnElement`, as well as within the bases of ORM elements
1248 such as :class:`.InstrumentedAttribute`, and may be used in :pep:`484`
1249 typing to indicate arguments or return values that should behave
1250 as column expressions.
1252 .. versionadded:: 2.0.0b4
1255 """
1257 __slots__ = ()
1260_SQO = SQLCoreOperations
1263class ColumnElement(
1264 roles.ColumnArgumentOrKeyRole,
1265 roles.StatementOptionRole,
1266 roles.WhereHavingRole,
1267 roles.BinaryElementRole[_T],
1268 roles.OrderByRole,
1269 roles.ColumnsClauseRole,
1270 roles.LimitOffsetRole,
1271 roles.DMLColumnRole,
1272 roles.DDLConstraintColumnRole,
1273 roles.DDLExpressionRole,
1274 SQLColumnExpression[_T],
1275 DQLDMLClauseElement,
1276):
1277 """Represent a column-oriented SQL expression suitable for usage in the
1278 "columns" clause, WHERE clause etc. of a statement.
1280 While the most familiar kind of :class:`_expression.ColumnElement` is the
1281 :class:`_schema.Column` object, :class:`_expression.ColumnElement`
1282 serves as the basis
1283 for any unit that may be present in a SQL expression, including
1284 the expressions themselves, SQL functions, bound parameters,
1285 literal expressions, keywords such as ``NULL``, etc.
1286 :class:`_expression.ColumnElement`
1287 is the ultimate base class for all such elements.
1289 A wide variety of SQLAlchemy Core functions work at the SQL expression
1290 level, and are intended to accept instances of
1291 :class:`_expression.ColumnElement` as
1292 arguments. These functions will typically document that they accept a
1293 "SQL expression" as an argument. What this means in terms of SQLAlchemy
1294 usually refers to an input which is either already in the form of a
1295 :class:`_expression.ColumnElement` object,
1296 or a value which can be **coerced** into
1297 one. The coercion rules followed by most, but not all, SQLAlchemy Core
1298 functions with regards to SQL expressions are as follows:
1300 * a literal Python value, such as a string, integer or floating
1301 point value, boolean, datetime, ``Decimal`` object, or virtually
1302 any other Python object, will be coerced into a "literal bound
1303 value". This generally means that a :func:`.bindparam` will be
1304 produced featuring the given value embedded into the construct; the
1305 resulting :class:`.BindParameter` object is an instance of
1306 :class:`_expression.ColumnElement`.
1307 The Python value will ultimately be sent
1308 to the DBAPI at execution time as a parameterized argument to the
1309 ``execute()`` or ``executemany()`` methods, after SQLAlchemy
1310 type-specific converters (e.g. those provided by any associated
1311 :class:`.TypeEngine` objects) are applied to the value.
1313 * any special object value, typically ORM-level constructs, which
1314 feature an accessor called ``__clause_element__()``. The Core
1315 expression system looks for this method when an object of otherwise
1316 unknown type is passed to a function that is looking to coerce the
1317 argument into a :class:`_expression.ColumnElement` and sometimes a
1318 :class:`_expression.SelectBase` expression.
1319 It is used within the ORM to
1320 convert from ORM-specific objects like mapped classes and
1321 mapped attributes into Core expression objects.
1323 * The Python ``None`` value is typically interpreted as ``NULL``,
1324 which in SQLAlchemy Core produces an instance of :func:`.null`.
1326 A :class:`_expression.ColumnElement` provides the ability to generate new
1327 :class:`_expression.ColumnElement`
1328 objects using Python expressions. This means that Python operators
1329 such as ``==``, ``!=`` and ``<`` are overloaded to mimic SQL operations,
1330 and allow the instantiation of further :class:`_expression.ColumnElement`
1331 instances
1332 which are composed from other, more fundamental
1333 :class:`_expression.ColumnElement`
1334 objects. For example, two :class:`.ColumnClause` objects can be added
1335 together with the addition operator ``+`` to produce
1336 a :class:`.BinaryExpression`.
1337 Both :class:`.ColumnClause` and :class:`.BinaryExpression` are subclasses
1338 of :class:`_expression.ColumnElement`:
1340 .. sourcecode:: pycon+sql
1342 >>> from sqlalchemy.sql import column
1343 >>> column("a") + column("b")
1344 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
1345 >>> print(column("a") + column("b"))
1346 {printsql}a + b
1348 .. seealso::
1350 :class:`_schema.Column`
1352 :func:`_expression.column`
1354 """
1356 __visit_name__ = "column_element"
1358 primary_key: bool = False
1359 _is_clone_of: Optional[ColumnElement[_T]]
1360 _is_column_element = True
1361 _insert_sentinel: bool = False
1362 _omit_from_statements = False
1363 _is_collection_aggregate = False
1365 foreign_keys: AbstractSet[ForeignKey] = frozenset()
1367 @util.memoized_property
1368 def _proxies(self) -> List[ColumnElement[Any]]:
1369 return []
1371 @util.non_memoized_property
1372 def _tq_label(self) -> Optional[str]:
1373 """The named label that can be used to target
1374 this column in a result set in a "table qualified" context.
1376 This label is almost always the label used when
1377 rendering <expr> AS <label> in a SELECT statement when using
1378 the LABEL_STYLE_TABLENAME_PLUS_COL label style, which is what the
1379 legacy ORM ``Query`` object uses as well.
1381 For a regular Column bound to a Table, this is typically the label
1382 <tablename>_<columnname>. For other constructs, different rules
1383 may apply, such as anonymized labels and others.
1385 .. versionchanged:: 1.4.21 renamed from ``._label``
1387 """
1388 return None
1390 key: Optional[str] = None
1391 """The 'key' that in some circumstances refers to this object in a
1392 Python namespace.
1394 This typically refers to the "key" of the column as present in the
1395 ``.c`` collection of a selectable, e.g. ``sometable.c["somekey"]`` would
1396 return a :class:`_schema.Column` with a ``.key`` of "somekey".
1398 """
1400 @HasMemoized.memoized_attribute
1401 def _tq_key_label(self) -> Optional[str]:
1402 """A label-based version of 'key' that in some circumstances refers
1403 to this object in a Python namespace.
1406 _tq_key_label comes into play when a select() statement is constructed
1407 with apply_labels(); in this case, all Column objects in the ``.c``
1408 collection are rendered as <tablename>_<columnname> in SQL; this is
1409 essentially the value of ._label. But to locate those columns in the
1410 ``.c`` collection, the name is along the lines of <tablename>_<key>;
1411 that's the typical value of .key_label.
1413 .. versionchanged:: 1.4.21 renamed from ``._key_label``
1415 """
1416 return self._proxy_key
1418 @property
1419 def _key_label(self) -> Optional[str]:
1420 """legacy; renamed to _tq_key_label"""
1421 return self._tq_key_label
1423 @property
1424 def _label(self) -> Optional[str]:
1425 """legacy; renamed to _tq_label"""
1426 return self._tq_label
1428 @property
1429 def _non_anon_label(self) -> Optional[str]:
1430 """the 'name' that naturally applies this element when rendered in
1431 SQL.
1433 Concretely, this is the "name" of a column or a label in a
1434 SELECT statement; ``<columnname>`` and ``<labelname>`` below:
1436 .. sourcecode:: sql
1438 SELECT <columnmame> FROM table
1440 SELECT column AS <labelname> FROM table
1442 Above, the two names noted will be what's present in the DBAPI
1443 ``cursor.description`` as the names.
1445 If this attribute returns ``None``, it means that the SQL element as
1446 written does not have a 100% fully predictable "name" that would appear
1447 in the ``cursor.description``. Examples include SQL functions, CAST
1448 functions, etc. While such things do return names in
1449 ``cursor.description``, they are only predictable on a
1450 database-specific basis; e.g. an expression like ``MAX(table.col)`` may
1451 appear as the string ``max`` on one database (like PostgreSQL) or may
1452 appear as the whole expression ``max(table.col)`` on SQLite.
1454 The default implementation looks for a ``.name`` attribute on the
1455 object, as has been the precedent established in SQLAlchemy for many
1456 years. An exception is made on the ``FunctionElement`` subclass
1457 so that the return value is always ``None``.
1459 .. versionadded:: 1.4.21
1463 """
1464 return getattr(self, "name", None)
1466 _render_label_in_columns_clause = True
1467 """A flag used by select._columns_plus_names that helps to determine
1468 we are actually going to render in terms of "SELECT <col> AS <label>".
1469 This flag can be returned as False for some Column objects that want
1470 to be rendered as simple "SELECT <col>"; typically columns that don't have
1471 any parent table and are named the same as what the label would be
1472 in any case.
1474 """
1476 _allow_label_resolve = True
1477 """A flag that can be flipped to prevent a column from being resolvable
1478 by string label name.
1480 The joined eager loader strategy in the ORM uses this, for example.
1482 """
1484 _is_implicitly_boolean = False
1486 _alt_names: Sequence[str] = ()
1488 if TYPE_CHECKING:
1490 def _ungroup(self) -> ColumnElement[_T]: ...
1492 @overload
1493 def self_group(self, against: None = None) -> ColumnElement[_T]: ...
1495 @overload
1496 def self_group(
1497 self, against: Optional[OperatorType] = None
1498 ) -> ColumnElement[Any]: ...
1500 def self_group(
1501 self, against: Optional[OperatorType] = None
1502 ) -> ColumnElement[Any]:
1503 if (
1504 against in (operators.and_, operators.or_, operators._asbool)
1505 and self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity
1506 ):
1507 return AsBoolean(self, operators.is_true, operators.is_false)
1508 elif against in (operators.any_op, operators.all_op):
1509 return Grouping(self)
1510 else:
1511 return self
1513 @overload
1514 def _negate(self: ColumnElement[bool]) -> ColumnElement[bool]: ...
1516 @overload
1517 def _negate(self: ColumnElement[_T]) -> ColumnElement[_T]: ...
1519 def _negate(self) -> ColumnElement[Any]:
1520 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
1521 return AsBoolean(self, operators.is_false, operators.is_true)
1522 else:
1523 grouped = self.self_group(against=operators.inv)
1524 assert isinstance(grouped, ColumnElement)
1525 return UnaryExpression(
1526 grouped,
1527 operator=operators.inv,
1528 )
1530 type: TypeEngine[_T]
1532 if not TYPE_CHECKING:
1534 @util.memoized_property
1535 def type(self) -> TypeEngine[_T]: # noqa: A001
1536 # used for delayed setup of
1537 # type_api
1538 return type_api.NULLTYPE
1540 @HasMemoized.memoized_attribute
1541 def comparator(self) -> TypeEngine.Comparator[_T]:
1542 try:
1543 comparator_factory = self.type.comparator_factory
1544 except AttributeError as err:
1545 raise TypeError(
1546 "Object %r associated with '.type' attribute "
1547 "is not a TypeEngine class or object" % self.type
1548 ) from err
1549 else:
1550 return comparator_factory(self)
1552 def __setstate__(self, state):
1553 self.__dict__.update(state)
1555 def __getattr__(self, key: str) -> Any:
1556 try:
1557 return getattr(self.comparator, key)
1558 except AttributeError as err:
1559 raise AttributeError(
1560 "Neither %r object nor %r object has an attribute %r"
1561 % (
1562 type(self).__name__,
1563 type(self.comparator).__name__,
1564 key,
1565 )
1566 ) from err
1568 def operate(
1569 self,
1570 op: operators.OperatorType,
1571 *other: Any,
1572 **kwargs: Any,
1573 ) -> ColumnElement[Any]:
1574 return op(self.comparator, *other, **kwargs) # type: ignore[no-any-return] # noqa: E501
1576 def reverse_operate(
1577 self, op: operators.OperatorType, other: Any, **kwargs: Any
1578 ) -> ColumnElement[Any]:
1579 return op(other, self.comparator, **kwargs) # type: ignore[no-any-return] # noqa: E501
1581 def _bind_param(
1582 self,
1583 operator: operators.OperatorType,
1584 obj: Any,
1585 type_: Optional[TypeEngine[_T]] = None,
1586 expanding: bool = False,
1587 ) -> BindParameter[_T]:
1588 return BindParameter(
1589 None,
1590 obj,
1591 _compared_to_operator=operator,
1592 type_=type_,
1593 _compared_to_type=self.type,
1594 unique=True,
1595 expanding=expanding,
1596 )
1598 @property
1599 def expression(self) -> ColumnElement[Any]:
1600 """Return a column expression.
1602 Part of the inspection interface; returns self.
1604 """
1605 return self
1607 @property
1608 def _select_iterable(self) -> _SelectIterable:
1609 return (self,)
1611 @util.memoized_property
1612 def base_columns(self) -> FrozenSet[ColumnElement[Any]]:
1613 return frozenset(c for c in self.proxy_set if not c._proxies)
1615 @util.memoized_property
1616 def proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1617 """set of all columns we are proxying
1619 as of 2.0 this is explicitly deannotated columns. previously it was
1620 effectively deannotated columns but wasn't enforced. annotated
1621 columns should basically not go into sets if at all possible because
1622 their hashing behavior is very non-performant.
1624 """
1625 return frozenset([self._deannotate()]).union(
1626 itertools.chain(*[c.proxy_set for c in self._proxies])
1627 )
1629 @util.memoized_property
1630 def _expanded_proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1631 return frozenset(_expand_cloned(self.proxy_set))
1633 def _uncached_proxy_list(self) -> List[ColumnElement[Any]]:
1634 """An 'uncached' version of proxy set.
1636 This list includes annotated columns which perform very poorly in
1637 set operations.
1639 """
1641 return [self] + list(
1642 itertools.chain(*[c._uncached_proxy_list() for c in self._proxies])
1643 )
1645 def shares_lineage(self, othercolumn: ColumnElement[Any]) -> bool:
1646 """Return True if the given :class:`_expression.ColumnElement`
1647 has a common ancestor to this :class:`_expression.ColumnElement`."""
1649 return bool(self.proxy_set.intersection(othercolumn.proxy_set))
1651 def _compare_name_for_result(self, other: ColumnElement[Any]) -> bool:
1652 """Return True if the given column element compares to this one
1653 when targeting within a result row."""
1655 return (
1656 hasattr(other, "name")
1657 and hasattr(self, "name")
1658 and other.name == self.name
1659 )
1661 @HasMemoized.memoized_attribute
1662 def _proxy_key(self) -> Optional[str]:
1663 if self._annotations and "proxy_key" in self._annotations:
1664 return cast(str, self._annotations["proxy_key"])
1666 name = self.key
1667 if not name:
1668 # there's a bit of a seeming contradiction which is that the
1669 # "_non_anon_label" of a column can in fact be an
1670 # "_anonymous_label"; this is when it's on a column that is
1671 # proxying for an anonymous expression in a subquery.
1672 name = self._non_anon_label
1674 if isinstance(name, _anonymous_label):
1675 return None
1676 else:
1677 return name
1679 @HasMemoized.memoized_attribute
1680 def _expression_label(self) -> Optional[str]:
1681 """a suggested label to use in the case that the column has no name,
1682 which should be used if possible as the explicit 'AS <label>'
1683 where this expression would normally have an anon label.
1685 this is essentially mostly what _proxy_key does except it returns
1686 None if the column has a normal name that can be used.
1688 """
1690 if getattr(self, "name", None) is not None:
1691 return None
1692 elif self._annotations and "proxy_key" in self._annotations:
1693 return cast(str, self._annotations["proxy_key"])
1694 else:
1695 return None
1697 def _make_proxy(
1698 self,
1699 selectable: FromClause,
1700 *,
1701 primary_key: ColumnSet,
1702 foreign_keys: Set[KeyedColumnElement[Any]],
1703 name: Optional[str] = None,
1704 key: Optional[str] = None,
1705 name_is_truncatable: bool = False,
1706 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
1707 **kw: Any,
1708 ) -> typing_Tuple[str, ColumnClause[_T]]:
1709 """Create a new :class:`_expression.ColumnElement` representing this
1710 :class:`_expression.ColumnElement` as it appears in the select list of
1711 a descending selectable.
1713 """
1714 if name is None:
1715 name = self._anon_name_label
1716 if key is None:
1717 key = self._proxy_key
1718 else:
1719 key = name
1721 assert key is not None
1723 co: ColumnClause[_T] = ColumnClause(
1724 (
1725 coercions.expect(roles.TruncatedLabelRole, name)
1726 if name_is_truncatable
1727 else name
1728 ),
1729 type_=getattr(self, "type", None),
1730 _selectable=selectable,
1731 )
1733 co._propagate_attrs = selectable._propagate_attrs
1734 if compound_select_cols:
1735 co._proxies = list(compound_select_cols)
1736 else:
1737 co._proxies = [self]
1738 if selectable._is_clone_of is not None:
1739 co._is_clone_of = selectable._is_clone_of.columns.get(key)
1740 return key, co
1742 def cast(self, type_: _TypeEngineArgument[_OPT]) -> Cast[_OPT]:
1743 """Produce a type cast, i.e. ``CAST(<expression> AS <type>)``.
1745 This is a shortcut to the :func:`_expression.cast` function.
1747 .. seealso::
1749 :ref:`tutorial_casts`
1751 :func:`_expression.cast`
1753 :func:`_expression.type_coerce`
1755 """
1756 return Cast(self, type_)
1758 def label(self, name: Optional[str]) -> Label[_T]:
1759 """Produce a column label, i.e. ``<columnname> AS <name>``.
1761 This is a shortcut to the :func:`_expression.label` function.
1763 If 'name' is ``None``, an anonymous label name will be generated.
1765 """
1766 return Label(name, self, self.type)
1768 def _anon_label(
1769 self, seed: Optional[str], add_hash: Optional[int] = None
1770 ) -> _anonymous_label:
1771 while self._is_clone_of is not None:
1772 self = self._is_clone_of
1774 # as of 1.4 anonymous label for ColumnElement uses hash(), not id(),
1775 # as the identifier, because a column and its annotated version are
1776 # the same thing in a SQL statement
1777 hash_value = hash(self)
1779 if add_hash:
1780 # this path is used for disambiguating anon labels that would
1781 # otherwise be the same name for the same element repeated.
1782 # an additional numeric value is factored in for each label.
1784 # shift hash(self) (which is id(self), typically 8 byte integer)
1785 # 16 bits leftward. fill extra add_hash on right
1786 assert add_hash < (2 << 15)
1787 assert seed
1788 hash_value = (hash_value << 16) | add_hash
1790 # extra underscore is added for labels with extra hash
1791 # values, to isolate the "deduped anon" namespace from the
1792 # regular namespace. eliminates chance of these
1793 # manufactured hash values overlapping with regular ones for some
1794 # undefined python interpreter
1795 seed = seed + "_"
1797 if isinstance(seed, _anonymous_label):
1798 # NOTE: the space after the hash is required
1799 return _anonymous_label(f"{seed}%({hash_value} )s")
1801 return _anonymous_label.safe_construct(hash_value, seed or "anon")
1803 @util.memoized_property
1804 def _anon_name_label(self) -> str:
1805 """Provides a constant 'anonymous label' for this ColumnElement.
1807 This is a label() expression which will be named at compile time.
1808 The same label() is returned each time ``anon_label`` is called so
1809 that expressions can reference ``anon_label`` multiple times,
1810 producing the same label name at compile time.
1812 The compiler uses this function automatically at compile time
1813 for expressions that are known to be 'unnamed' like binary
1814 expressions and function calls.
1816 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1817 public and is renamed to _anon_name_label. anon_name exists
1818 for backwards compat
1820 """
1821 name = getattr(self, "name", None)
1822 return self._anon_label(name)
1824 @util.memoized_property
1825 def _anon_key_label(self) -> _anonymous_label:
1826 """Provides a constant 'anonymous key label' for this ColumnElement.
1828 Compare to ``anon_label``, except that the "key" of the column,
1829 if available, is used to generate the label.
1831 This is used when a deduplicating key is placed into the columns
1832 collection of a selectable.
1834 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1835 public and is renamed to _anon_key_label. anon_key_label exists
1836 for backwards compat
1838 """
1839 return self._anon_label(self._proxy_key)
1841 @property
1842 @util.deprecated(
1843 "1.4",
1844 "The :attr:`_expression.ColumnElement.anon_label` attribute is now "
1845 "private, and the public accessor is deprecated.",
1846 )
1847 def anon_label(self) -> str:
1848 return self._anon_name_label
1850 @property
1851 @util.deprecated(
1852 "1.4",
1853 "The :attr:`_expression.ColumnElement.anon_key_label` attribute is "
1854 "now private, and the public accessor is deprecated.",
1855 )
1856 def anon_key_label(self) -> str:
1857 return self._anon_key_label
1859 def _dedupe_anon_label_idx(self, idx: int) -> str:
1860 """label to apply to a column that is anon labeled, but repeated
1861 in the SELECT, so that we have to make an "extra anon" label that
1862 disambiguates it from the previous appearance.
1864 these labels come out like "foo_bar_id__1" and have double underscores
1865 in them.
1867 """
1868 label = getattr(self, "name", None)
1870 # current convention is that if the element doesn't have a
1871 # ".name" (usually because it is not NamedColumn), we try to
1872 # use a "table qualified" form for the "dedupe anon" label,
1873 # based on the notion that a label like
1874 # "CAST(casttest.v1 AS DECIMAL) AS casttest_v1__1" looks better than
1875 # "CAST(casttest.v1 AS DECIMAL) AS anon__1"
1877 if label is None:
1878 return self._dedupe_anon_tq_label_idx(idx)
1879 else:
1880 return self._anon_label(label, add_hash=idx)
1882 @util.memoized_property
1883 def _anon_tq_label(self) -> _anonymous_label:
1884 return self._anon_label(getattr(self, "_tq_label", None))
1886 @util.memoized_property
1887 def _anon_tq_key_label(self) -> _anonymous_label:
1888 return self._anon_label(getattr(self, "_tq_key_label", None))
1890 def _dedupe_anon_tq_label_idx(self, idx: int) -> _anonymous_label:
1891 label = getattr(self, "_tq_label", None) or "anon"
1893 return self._anon_label(label, add_hash=idx)
1896class KeyedColumnElement(ColumnElement[_T]):
1897 """ColumnElement where ``.key`` is non-None."""
1899 _is_keyed_column_element = True
1901 key: str
1904class WrapsColumnExpression(ColumnElement[_T]):
1905 """Mixin that defines a :class:`_expression.ColumnElement`
1906 as a wrapper with special
1907 labeling behavior for an expression that already has a name.
1909 .. versionadded:: 1.4
1911 .. seealso::
1913 :ref:`change_4449`
1916 """
1918 @property
1919 def wrapped_column_expression(self) -> ColumnElement[_T]:
1920 raise NotImplementedError()
1922 @util.non_memoized_property
1923 def _tq_label(self) -> Optional[str]:
1924 wce = self.wrapped_column_expression
1925 if hasattr(wce, "_tq_label"):
1926 return wce._tq_label
1927 else:
1928 return None
1930 @property
1931 def _label(self) -> Optional[str]:
1932 return self._tq_label
1934 @property
1935 def _non_anon_label(self) -> Optional[str]:
1936 return None
1938 @util.non_memoized_property
1939 def _anon_name_label(self) -> str:
1940 wce = self.wrapped_column_expression
1942 # this logic tries to get the WrappedColumnExpression to render
1943 # with "<expr> AS <name>", where "<name>" is the natural name
1944 # within the expression itself. e.g. "CAST(table.foo) AS foo".
1945 if not wce._is_text_clause:
1946 nal = wce._non_anon_label
1947 if nal:
1948 return nal
1949 elif hasattr(wce, "_anon_name_label"):
1950 return wce._anon_name_label
1951 return super()._anon_name_label
1953 def _dedupe_anon_label_idx(self, idx: int) -> str:
1954 wce = self.wrapped_column_expression
1955 nal = wce._non_anon_label
1956 if nal:
1957 return self._anon_label(nal + "_")
1958 else:
1959 return self._dedupe_anon_tq_label_idx(idx)
1961 @property
1962 def _proxy_key(self):
1963 wce = self.wrapped_column_expression
1965 if not wce._is_text_clause:
1966 return wce._proxy_key
1967 return super()._proxy_key
1970class DMLTargetCopy(roles.InElementRole, KeyedColumnElement[_T]):
1971 """Refer to another column's VALUES or SET expression in an INSERT or
1972 UPDATE statement.
1974 See the public-facing :func:`_sql.from_dml_column` constructor for
1975 background.
1977 .. versionadded:: 2.1
1980 """
1982 def __init__(self, column: _DMLOnlyColumnArgument[_T]):
1983 self.column = coercions.expect(roles.ColumnArgumentRole, column)
1984 self.type = self.column.type
1986 __visit_name__ = "dmltargetcopy"
1988 _traverse_internals: _TraverseInternalsType = [
1989 ("column", InternalTraversal.dp_clauseelement),
1990 ]
1993class BindParameter(roles.InElementRole, KeyedColumnElement[_T]):
1994 r"""Represent a "bound expression".
1996 :class:`.BindParameter` is invoked explicitly using the
1997 :func:`.bindparam` function, as in::
1999 from sqlalchemy import bindparam
2001 stmt = select(users_table).where(
2002 users_table.c.name == bindparam("username")
2003 )
2005 Detailed discussion of how :class:`.BindParameter` is used is
2006 at :func:`.bindparam`.
2008 .. seealso::
2010 :func:`.bindparam`
2012 """
2014 __visit_name__ = "bindparam"
2016 _traverse_internals: _TraverseInternalsType = [
2017 ("key", InternalTraversal.dp_anon_name),
2018 ("type", InternalTraversal.dp_type),
2019 ("callable", InternalTraversal.dp_plain_dict),
2020 ("value", InternalTraversal.dp_plain_obj),
2021 ("literal_execute", InternalTraversal.dp_boolean),
2022 ]
2024 key: str
2025 _anon_map_key: Optional[str] = None
2026 type: TypeEngine[_T]
2027 value: Optional[_T]
2029 _is_crud = False
2030 _is_bind_parameter = True
2032 # bindparam implements its own _gen_cache_key() method however
2033 # we check subclasses for this flag, else no cache key is generated
2034 inherit_cache = True
2036 def __init__(
2037 self,
2038 key: Optional[str],
2039 value: Any = _NoArg.NO_ARG,
2040 type_: Optional[_TypeEngineArgument[_T]] = None,
2041 unique: bool = False,
2042 required: Union[bool, Literal[_NoArg.NO_ARG]] = _NoArg.NO_ARG,
2043 quote: Optional[bool] = None,
2044 callable_: Optional[Callable[[], Any]] = None,
2045 expanding: bool = False,
2046 isoutparam: bool = False,
2047 literal_execute: bool = False,
2048 _compared_to_operator: Optional[OperatorType] = None,
2049 _compared_to_type: Optional[TypeEngine[Any]] = None,
2050 _is_crud: bool = False,
2051 ):
2052 if required is _NoArg.NO_ARG:
2053 required = value is _NoArg.NO_ARG and callable_ is None
2054 if value is _NoArg.NO_ARG:
2055 value = None
2057 if quote is not None:
2058 key = quoted_name.construct(key, quote)
2060 if unique:
2061 self.key, self._anon_map_key = (
2062 _anonymous_label.safe_construct_with_key(
2063 id(self),
2064 (
2065 key
2066 if key is not None
2067 and not isinstance(key, _anonymous_label)
2068 else "param"
2069 ),
2070 sanitize_key=True,
2071 )
2072 )
2073 elif key:
2074 self.key = key
2075 else:
2076 self.key, self._anon_map_key = (
2077 _anonymous_label.safe_construct_with_key(id(self), "param")
2078 )
2080 # identifying key that won't change across
2081 # clones, used to identify the bind's logical
2082 # identity
2083 self._identifying_key = self.key
2085 # key that was passed in the first place, used to
2086 # generate new keys
2087 self._orig_key = key or "param"
2089 self.unique = unique
2090 self.value = value
2091 self.callable = callable_
2092 self.isoutparam = isoutparam
2093 self.required = required
2095 # indicate an "expanding" parameter; the compiler sets this
2096 # automatically in the compiler _render_in_expr_w_bindparam method
2097 # for an IN expression
2098 self.expanding = expanding
2100 # this is another hint to help w/ expanding and is typically
2101 # set in the compiler _render_in_expr_w_bindparam method for an
2102 # IN expression
2103 self.expand_op = None
2105 self.literal_execute = literal_execute
2106 if _is_crud:
2107 self._is_crud = True
2109 if type_ is None:
2110 if expanding:
2111 if value:
2112 check_value = value[0]
2113 else:
2114 check_value = type_api._NO_VALUE_IN_LIST
2115 else:
2116 check_value = value
2117 if _compared_to_type is not None:
2118 self.type = _compared_to_type.coerce_compared_value(
2119 _compared_to_operator, check_value
2120 )
2121 else:
2122 self.type = type_api._resolve_value_to_type(check_value)
2123 elif isinstance(type_, type):
2124 self.type = type_()
2125 elif is_tuple_type(type_):
2126 if value:
2127 if expanding:
2128 check_value = value[0]
2129 else:
2130 check_value = value
2131 cast("BindParameter[TupleAny]", self).type = (
2132 type_._resolve_values_to_types(check_value)
2133 )
2134 else:
2135 cast("BindParameter[TupleAny]", self).type = type_
2136 else:
2137 self.type = type_
2139 def _with_value(self, value, maintain_key=False, required=NO_ARG):
2140 """Return a copy of this :class:`.BindParameter` with the given value
2141 set.
2142 """
2143 cloned = self._clone(maintain_key=maintain_key)
2144 cloned.value = value
2145 cloned.callable = None
2146 cloned.required = required if required is not NO_ARG else self.required
2147 if cloned.type is type_api.NULLTYPE:
2148 cloned.type = type_api._resolve_value_to_type(value)
2149 return cloned
2151 @property
2152 def effective_value(self) -> Optional[_T]:
2153 """Return the value of this bound parameter,
2154 taking into account if the ``callable`` parameter
2155 was set.
2157 The ``callable`` value will be evaluated
2158 and returned if present, else ``value``.
2160 """
2161 if self.callable:
2162 # TODO: set up protocol for bind parameter callable
2163 return self.callable() # type: ignore
2164 else:
2165 return self.value
2167 def render_literal_execute(self) -> Self:
2168 """Produce a copy of this bound parameter that will enable the
2169 :paramref:`_sql.BindParameter.literal_execute` flag.
2171 The :paramref:`_sql.BindParameter.literal_execute` flag will
2172 have the effect of the parameter rendered in the compiled SQL
2173 string using ``[POSTCOMPILE]`` form, which is a special form that
2174 is converted to be a rendering of the literal value of the parameter
2175 at SQL execution time. The rationale is to support caching
2176 of SQL statement strings that can embed per-statement literal values,
2177 such as LIMIT and OFFSET parameters, in the final SQL string that
2178 is passed to the DBAPI. Dialects in particular may want to use
2179 this method within custom compilation schemes.
2181 .. versionadded:: 1.4.5
2183 .. seealso::
2185 :ref:`engine_thirdparty_caching`
2187 """
2188 c: Self = ClauseElement._clone(self)
2189 c.literal_execute = True
2190 return c
2192 def _negate_in_binary(self, negated_op, original_op):
2193 if self.expand_op is original_op:
2194 bind = self._clone()
2195 bind.expand_op = negated_op
2196 return bind
2197 else:
2198 return self
2200 def _with_binary_element_type(self, type_: TypeEngine[Any]) -> Self:
2201 c: Self = ClauseElement._clone(self)
2202 c.type = type_
2203 return c
2205 def _clone(self, maintain_key: bool = False, **kw: Any) -> Self:
2206 c: Self = ClauseElement._clone(self, **kw)
2207 # ensure all the BindParameter objects stay in cloned set.
2208 # in #7823, we changed "clone" so that a clone only keeps a reference
2209 # to the "original" element, since for column correspondence, that's
2210 # all we need. However, for BindParam, _cloned_set is used by
2211 # the "cache key bind match" lookup, which means if any of those
2212 # interim BindParameter objects became part of a cache key in the
2213 # cache, we need it. So here, make sure all clones keep carrying
2214 # forward.
2215 c._cloned_set.update(self._cloned_set)
2216 if not maintain_key and self.unique:
2217 c.key, c._anon_map_key = _anonymous_label.safe_construct_with_key(
2218 id(c), c._orig_key or "param", sanitize_key=True
2219 )
2220 return c
2222 def _gen_cache_key(self, anon_map, bindparams):
2223 _gen_cache_ok = self.__class__.__dict__.get("inherit_cache", False)
2225 if not _gen_cache_ok:
2226 if anon_map is not None:
2227 anon_map[NO_CACHE] = True
2228 return None
2230 id_, found = anon_map.get_anon(self)
2231 if found:
2232 return (id_, self.__class__)
2234 if bindparams is not None:
2235 bindparams.append(self)
2237 return (
2238 id_,
2239 self.__class__,
2240 self.type._static_cache_key,
2241 (
2242 anon_map[self._anon_map_key]
2243 if self._anon_map_key is not None
2244 else self.key
2245 ),
2246 self.literal_execute,
2247 )
2249 def _convert_to_unique(self):
2250 if not self.unique:
2251 self.unique = True
2252 self.key, self._anon_map_key = (
2253 _anonymous_label.safe_construct_with_key(
2254 id(self), self._orig_key or "param", sanitize_key=True
2255 )
2256 )
2258 def __getstate__(self):
2259 """execute a deferred value for serialization purposes."""
2261 d = self.__dict__.copy()
2262 v = self.value
2263 if self.callable:
2264 v = self.callable()
2265 d["callable"] = None
2266 d["value"] = v
2267 return d
2269 def __setstate__(self, state):
2270 if state.get("unique", False):
2271 anon_and_key = _anonymous_label.safe_construct_with_key(
2272 id(self), state.get("_orig_key", "param"), sanitize_key=True
2273 )
2274 state["key"], state["_anon_map_key"] = anon_and_key
2275 self.__dict__.update(state)
2277 def __repr__(self):
2278 return "%s(%r, %r, type_=%r)" % (
2279 self.__class__.__name__,
2280 self.key,
2281 self.value,
2282 self.type,
2283 )
2286class TypeClause(DQLDMLClauseElement):
2287 """Handle a type keyword in a SQL statement.
2289 Used by the ``Case`` statement.
2291 """
2293 __visit_name__ = "typeclause"
2295 _traverse_internals: _TraverseInternalsType = [
2296 ("type", InternalTraversal.dp_type)
2297 ]
2298 type: TypeEngine[Any]
2300 def __init__(self, type_: TypeEngine[Any]):
2301 self.type = type_
2304class TextClause(
2305 roles.DDLConstraintColumnRole,
2306 roles.DDLExpressionRole,
2307 roles.StatementOptionRole,
2308 roles.WhereHavingRole,
2309 roles.OrderByRole,
2310 roles.FromClauseRole,
2311 roles.SelectStatementRole,
2312 roles.InElementRole,
2313 Generative,
2314 Executable,
2315 DQLDMLClauseElement,
2316 roles.BinaryElementRole[Any],
2317 inspection.Inspectable["TextClause"],
2318):
2319 """Represent a literal SQL text fragment.
2321 E.g.::
2323 from sqlalchemy import text
2325 t = text("SELECT * FROM users")
2326 result = connection.execute(t)
2328 The :class:`_expression.TextClause` construct is produced using the
2329 :func:`_expression.text`
2330 function; see that function for full documentation.
2332 .. seealso::
2334 :func:`_expression.text`
2336 """
2338 __visit_name__ = "textclause"
2340 _traverse_internals: _TraverseInternalsType = [
2341 ("_bindparams", InternalTraversal.dp_string_clauseelement_dict),
2342 ("text", InternalTraversal.dp_string),
2343 ] + Executable._executable_traverse_internals
2345 _is_text_clause = True
2347 _is_textual = True
2349 _bind_params_regex = re.compile(r"(?<![:\w\x5c]):(\w+)(?!:)", re.UNICODE)
2350 _is_implicitly_boolean = False
2352 _render_label_in_columns_clause = False
2354 _omit_from_statements = False
2356 _is_collection_aggregate = False
2358 @property
2359 def _hide_froms(self) -> Iterable[FromClause]:
2360 return ()
2362 def __and__(self, other):
2363 # support use in select.where(), query.filter()
2364 return and_(self, other)
2366 @property
2367 def _select_iterable(self) -> _SelectIterable:
2368 return (self,)
2370 # help in those cases where text() is
2371 # interpreted in a column expression situation
2372 key: Optional[str] = None
2373 _label: Optional[str] = None
2375 _allow_label_resolve = False
2377 @property
2378 def _is_star(self): # type: ignore[override]
2379 return self.text == "*"
2381 def __init__(self, text: str):
2382 self._bindparams: Dict[str, BindParameter[Any]] = {}
2384 def repl(m):
2385 self._bindparams[m.group(1)] = BindParameter(m.group(1))
2386 return ":%s" % m.group(1)
2388 # scan the string and search for bind parameter names, add them
2389 # to the list of bindparams
2390 self.text = self._bind_params_regex.sub(repl, text)
2392 @_generative
2393 def bindparams(
2394 self,
2395 *binds: BindParameter[Any],
2396 **names_to_values: Any,
2397 ) -> Self:
2398 """Establish the values and/or types of bound parameters within
2399 this :class:`_expression.TextClause` construct.
2401 Given a text construct such as::
2403 from sqlalchemy import text
2405 stmt = text(
2406 "SELECT id, name FROM user WHERE name=:name AND timestamp=:timestamp"
2407 )
2409 the :meth:`_expression.TextClause.bindparams`
2410 method can be used to establish
2411 the initial value of ``:name`` and ``:timestamp``,
2412 using simple keyword arguments::
2414 stmt = stmt.bindparams(
2415 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2416 )
2418 Where above, new :class:`.BindParameter` objects
2419 will be generated with the names ``name`` and ``timestamp``, and
2420 values of ``jack`` and ``datetime.datetime(2012, 10, 8, 15, 12, 5)``,
2421 respectively. The types will be
2422 inferred from the values given, in this case :class:`.String` and
2423 :class:`.DateTime`.
2425 When specific typing behavior is needed, the positional ``*binds``
2426 argument can be used in which to specify :func:`.bindparam` constructs
2427 directly. These constructs must include at least the ``key``
2428 argument, then an optional value and type::
2430 from sqlalchemy import bindparam
2432 stmt = stmt.bindparams(
2433 bindparam("name", value="jack", type_=String),
2434 bindparam("timestamp", type_=DateTime),
2435 )
2437 Above, we specified the type of :class:`.DateTime` for the
2438 ``timestamp`` bind, and the type of :class:`.String` for the ``name``
2439 bind. In the case of ``name`` we also set the default value of
2440 ``"jack"``.
2442 Additional bound parameters can be supplied at statement execution
2443 time, e.g.::
2445 result = connection.execute(
2446 stmt, timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2447 )
2449 The :meth:`_expression.TextClause.bindparams`
2450 method can be called repeatedly,
2451 where it will re-use existing :class:`.BindParameter` objects to add
2452 new information. For example, we can call
2453 :meth:`_expression.TextClause.bindparams`
2454 first with typing information, and a
2455 second time with value information, and it will be combined::
2457 stmt = text(
2458 "SELECT id, name FROM user WHERE name=:name "
2459 "AND timestamp=:timestamp"
2460 )
2461 stmt = stmt.bindparams(
2462 bindparam("name", type_=String), bindparam("timestamp", type_=DateTime)
2463 )
2464 stmt = stmt.bindparams(
2465 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2466 )
2468 The :meth:`_expression.TextClause.bindparams`
2469 method also supports the concept of
2470 **unique** bound parameters. These are parameters that are
2471 "uniquified" on name at statement compilation time, so that multiple
2472 :func:`_expression.text`
2473 constructs may be combined together without the names
2474 conflicting. To use this feature, specify the
2475 :paramref:`.BindParameter.unique` flag on each :func:`.bindparam`
2476 object::
2478 stmt1 = text("select id from table where name=:name").bindparams(
2479 bindparam("name", value="name1", unique=True)
2480 )
2481 stmt2 = text("select id from table where name=:name").bindparams(
2482 bindparam("name", value="name2", unique=True)
2483 )
2485 union = union_all(stmt1.columns(column("id")), stmt2.columns(column("id")))
2487 The above statement will render as:
2489 .. sourcecode:: sql
2491 select id from table where name=:name_1
2492 UNION ALL select id from table where name=:name_2
2494 """ # noqa: E501
2495 self._bindparams = new_params = self._bindparams.copy()
2497 for bind in binds:
2498 try:
2499 # the regex used for text() currently will not match
2500 # a unique/anonymous key in any case, so use the _orig_key
2501 # so that a text() construct can support unique parameters
2502 existing = new_params[bind._orig_key]
2503 except KeyError as err:
2504 raise exc.ArgumentError(
2505 "This text() construct doesn't define a "
2506 "bound parameter named %r" % bind._orig_key
2507 ) from err
2508 else:
2509 new_params[existing._orig_key] = bind
2511 for key, value in names_to_values.items():
2512 try:
2513 existing = new_params[key]
2514 except KeyError as err:
2515 raise exc.ArgumentError(
2516 "This text() construct doesn't define a "
2517 "bound parameter named %r" % key
2518 ) from err
2519 else:
2520 new_params[key] = existing._with_value(value, required=False)
2521 return self
2523 @util.preload_module("sqlalchemy.sql.selectable")
2524 def columns(
2525 self,
2526 *cols: _ColumnExpressionArgument[Any],
2527 **types: _TypeEngineArgument[Any],
2528 ) -> TextualSelect:
2529 r"""Turn this :class:`_expression.TextClause` object into a
2530 :class:`_expression.TextualSelect`
2531 object that serves the same role as a SELECT
2532 statement.
2534 The :class:`_expression.TextualSelect` is part of the
2535 :class:`_expression.SelectBase`
2536 hierarchy and can be embedded into another statement by using the
2537 :meth:`_expression.TextualSelect.subquery` method to produce a
2538 :class:`.Subquery`
2539 object, which can then be SELECTed from.
2541 This function essentially bridges the gap between an entirely
2542 textual SELECT statement and the SQL expression language concept
2543 of a "selectable"::
2545 from sqlalchemy.sql import column, text
2547 stmt = text("SELECT id, name FROM some_table")
2548 stmt = stmt.columns(column("id"), column("name")).subquery("st")
2550 stmt = (
2551 select(mytable)
2552 .select_from(mytable.join(stmt, mytable.c.name == stmt.c.name))
2553 .where(stmt.c.id > 5)
2554 )
2556 Above, we pass a series of :func:`_expression.column` elements to the
2557 :meth:`_expression.TextClause.columns` method positionally. These
2558 :func:`_expression.column`
2559 elements now become first class elements upon the
2560 :attr:`_expression.TextualSelect.selected_columns` column collection,
2561 which then
2562 become part of the :attr:`.Subquery.c` collection after
2563 :meth:`_expression.TextualSelect.subquery` is invoked.
2565 The column expressions we pass to
2566 :meth:`_expression.TextClause.columns` may
2567 also be typed; when we do so, these :class:`.TypeEngine` objects become
2568 the effective return type of the column, so that SQLAlchemy's
2569 result-set-processing systems may be used on the return values.
2570 This is often needed for types such as date or boolean types, as well
2571 as for unicode processing on some dialect configurations::
2573 stmt = text("SELECT id, name, timestamp FROM some_table")
2574 stmt = stmt.columns(
2575 column("id", Integer),
2576 column("name", Unicode),
2577 column("timestamp", DateTime),
2578 )
2580 for id, name, timestamp in connection.execute(stmt):
2581 print(id, name, timestamp)
2583 As a shortcut to the above syntax, keyword arguments referring to
2584 types alone may be used, if only type conversion is needed::
2586 stmt = text("SELECT id, name, timestamp FROM some_table")
2587 stmt = stmt.columns(id=Integer, name=Unicode, timestamp=DateTime)
2589 for id, name, timestamp in connection.execute(stmt):
2590 print(id, name, timestamp)
2592 The positional form of :meth:`_expression.TextClause.columns`
2593 also provides the
2594 unique feature of **positional column targeting**, which is
2595 particularly useful when using the ORM with complex textual queries. If
2596 we specify the columns from our model to
2597 :meth:`_expression.TextClause.columns`,
2598 the result set will match to those columns positionally, meaning the
2599 name or origin of the column in the textual SQL doesn't matter::
2601 stmt = text(
2602 "SELECT users.id, addresses.id, users.id, "
2603 "users.name, addresses.email_address AS email "
2604 "FROM users JOIN addresses ON users.id=addresses.user_id "
2605 "WHERE users.id = 1"
2606 ).columns(
2607 User.id,
2608 Address.id,
2609 Address.user_id,
2610 User.name,
2611 Address.email_address,
2612 )
2614 query = (
2615 session.query(User)
2616 .from_statement(stmt)
2617 .options(contains_eager(User.addresses))
2618 )
2620 The :meth:`_expression.TextClause.columns` method provides a direct
2621 route to calling :meth:`_expression.FromClause.subquery` as well as
2622 :meth:`_expression.SelectBase.cte`
2623 against a textual SELECT statement::
2625 stmt = stmt.columns(id=Integer, name=String).cte("st")
2627 stmt = select(sometable).where(sometable.c.id == stmt.c.id)
2629 :param \*cols: A series of :class:`_expression.ColumnElement` objects,
2630 typically
2631 :class:`_schema.Column` objects from a :class:`_schema.Table`
2632 or ORM level
2633 column-mapped attributes, representing a set of columns that this
2634 textual string will SELECT from.
2636 :param \**types: A mapping of string names to :class:`.TypeEngine`
2637 type objects indicating the datatypes to use for names that are
2638 SELECTed from the textual string. Prefer to use the ``*cols``
2639 argument as it also indicates positional ordering.
2641 """
2642 selectable = util.preloaded.sql_selectable
2644 input_cols: List[NamedColumn[Any]] = [
2645 coercions.expect(roles.LabeledColumnExprRole, col) for col in cols
2646 ]
2648 positional_input_cols = [
2649 (
2650 ColumnClause(col.key, types.pop(col.key))
2651 if col.key in types
2652 else col
2653 )
2654 for col in input_cols
2655 ]
2656 keyed_input_cols: List[NamedColumn[Any]] = [
2657 ColumnClause(key, type_) for key, type_ in types.items()
2658 ]
2660 elem = selectable.TextualSelect.__new__(selectable.TextualSelect)
2661 elem._init(
2662 self,
2663 positional_input_cols + keyed_input_cols,
2664 positional=bool(positional_input_cols) and not keyed_input_cols,
2665 )
2666 return elem
2668 @property
2669 def type(self) -> TypeEngine[Any]:
2670 return type_api.NULLTYPE
2672 @property
2673 def comparator(self):
2674 # TODO: this seems wrong, it seems like we might not
2675 # be using this method.
2676 return self.type.comparator_factory(self) # type: ignore
2678 def self_group(
2679 self, against: Optional[OperatorType] = None
2680 ) -> Union[Self, Grouping[Any]]:
2681 if against is operators.in_op:
2682 return Grouping(self)
2683 else:
2684 return self
2687class Null(SingletonConstant, roles.ConstExprRole[None], ColumnElement[None]):
2688 """Represent the NULL keyword in a SQL statement.
2690 :class:`.Null` is accessed as a constant via the
2691 :func:`.null` function.
2693 """
2695 __visit_name__ = "null"
2697 _traverse_internals: _TraverseInternalsType = []
2698 _singleton: Null
2700 if not TYPE_CHECKING:
2702 @util.memoized_property
2703 def type(self) -> TypeEngine[_T]: # noqa: A001
2704 return type_api.NULLTYPE
2706 @classmethod
2707 def _instance(cls) -> Null:
2708 """Return a constant :class:`.Null` construct."""
2710 return Null._singleton
2713Null._create_singleton()
2716class False_(
2717 SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]
2718):
2719 """Represent the ``false`` keyword, or equivalent, in a SQL statement.
2721 :class:`.False_` is accessed as a constant via the
2722 :func:`.false` function.
2724 """
2726 __visit_name__ = "false"
2727 _traverse_internals: _TraverseInternalsType = []
2728 _singleton: False_
2730 if not TYPE_CHECKING:
2732 @util.memoized_property
2733 def type(self) -> TypeEngine[_T]: # noqa: A001
2734 return type_api.BOOLEANTYPE
2736 def _negate(self) -> True_:
2737 return True_._singleton
2739 @classmethod
2740 def _instance(cls) -> False_:
2741 return False_._singleton
2744False_._create_singleton()
2747class True_(SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]):
2748 """Represent the ``true`` keyword, or equivalent, in a SQL statement.
2750 :class:`.True_` is accessed as a constant via the
2751 :func:`.true` function.
2753 """
2755 __visit_name__ = "true"
2757 _traverse_internals: _TraverseInternalsType = []
2758 _singleton: True_
2760 if not TYPE_CHECKING:
2762 @util.memoized_property
2763 def type(self) -> TypeEngine[_T]: # noqa: A001
2764 return type_api.BOOLEANTYPE
2766 def _negate(self) -> False_:
2767 return False_._singleton
2769 @classmethod
2770 def _ifnone(
2771 cls, other: Optional[ColumnElement[Any]]
2772 ) -> ColumnElement[Any]:
2773 if other is None:
2774 return cls._instance()
2775 else:
2776 return other
2778 @classmethod
2779 def _instance(cls) -> True_:
2780 return True_._singleton
2783True_._create_singleton()
2786class ElementList(DQLDMLClauseElement):
2787 """Describe a list of clauses that will be space separated.
2789 This is a minimal version of :class:`.ClauseList` which is used by
2790 the :class:`.HasSyntaxExtension` class. It does not do any coercions
2791 so should be used internally only.
2793 .. versionadded:: 2.1
2795 """
2797 __visit_name__ = "element_list"
2799 _traverse_internals: _TraverseInternalsType = [
2800 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2801 ]
2803 clauses: typing_Tuple[ClauseElement, ...]
2805 def __init__(self, clauses: Sequence[ClauseElement]):
2806 self.clauses = tuple(clauses)
2809class OrderByList(
2810 roles.OrderByRole,
2811 operators.OrderingOperators,
2812 DQLDMLClauseElement,
2813):
2814 """Describe a list of clauses that will be comma separated to nest
2815 within an ORDER BY.
2817 .. versionadded:: 2.1
2819 """
2821 __visit_name__ = "order_by_list"
2823 _traverse_internals: _TraverseInternalsType = [
2824 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2825 ]
2827 clauses: List[ColumnElement[Any]]
2829 def __init__(
2830 self,
2831 clauses: Iterable[Union[OrderByList, _ColumnExpressionArgument[Any]]],
2832 ):
2833 text_converter_role: Type[roles.SQLRole] = roles.ByOfRole
2834 self._text_converter_role = text_converter_role
2836 self.clauses = [
2837 coercions.expect(
2838 text_converter_role, clause, apply_propagate_attrs=self
2839 )
2840 for clause in clauses
2841 ]
2843 def __iter__(self) -> Iterator[ColumnElement[Any]]:
2844 return iter(self.clauses)
2846 def __len__(self) -> int:
2847 return len(self.clauses)
2849 @property
2850 def _select_iterable(self) -> _SelectIterable:
2851 return itertools.chain.from_iterable(
2852 [elem._select_iterable for elem in self.clauses]
2853 )
2855 @util.ro_non_memoized_property
2856 def _from_objects(self) -> List[FromClause]:
2857 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
2859 def self_group(
2860 self, against: Optional[OperatorType] = None
2861 ) -> Union[Self, Grouping[Any]]:
2862 return self
2864 def desc(self) -> OrderByList:
2865 return OrderByList([e.desc() for e in self.clauses])
2867 def asc(self) -> OrderByList:
2868 return OrderByList([e.asc() for e in self.clauses])
2870 def nulls_first(self) -> OrderByList:
2871 return OrderByList([e.nulls_first() for e in self.clauses])
2873 def nulls_last(self) -> OrderByList:
2874 return OrderByList([e.nulls_last() for e in self.clauses])
2877class ClauseList(
2878 roles.InElementRole,
2879 roles.OrderByRole,
2880 roles.ColumnsClauseRole,
2881 roles.DMLColumnRole,
2882 DQLDMLClauseElement,
2883):
2884 """Describe a list of clauses, separated by an operator.
2886 By default, is comma-separated, such as a column listing.
2888 """
2890 __visit_name__ = "clauselist"
2892 # Used by ORM context.py to identify ClauseList objects in legacy
2893 # composite attribute queries (see test_query_cols_legacy test)
2894 _is_clause_list = True
2896 _traverse_internals: _TraverseInternalsType = [
2897 ("clauses", InternalTraversal.dp_clauseelement_list),
2898 ("operator", InternalTraversal.dp_operator),
2899 ]
2901 clauses: List[ColumnElement[Any]]
2903 def __init__(
2904 self,
2905 *clauses: _ColumnExpressionArgument[Any],
2906 operator: OperatorType = operators.comma_op,
2907 group: bool = True,
2908 group_contents: bool = True,
2909 _literal_as_text_role: Type[roles.SQLRole] = roles.WhereHavingRole,
2910 ):
2911 self.operator = operator
2912 self.group = group
2913 self.group_contents = group_contents
2914 clauses_iterator: Iterable[_ColumnExpressionArgument[Any]] = clauses
2915 text_converter_role: Type[roles.SQLRole] = _literal_as_text_role
2916 self._text_converter_role = text_converter_role
2918 if self.group_contents:
2919 self.clauses = [
2920 coercions.expect(
2921 text_converter_role, clause, apply_propagate_attrs=self
2922 ).self_group(against=self.operator)
2923 for clause in clauses_iterator
2924 ]
2925 else:
2926 self.clauses = [
2927 coercions.expect(
2928 text_converter_role, clause, apply_propagate_attrs=self
2929 )
2930 for clause in clauses_iterator
2931 ]
2932 self._is_implicitly_boolean = operators.is_boolean(self.operator)
2934 @classmethod
2935 def _construct_raw(
2936 cls,
2937 operator: OperatorType,
2938 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
2939 ) -> ClauseList:
2940 self = cls.__new__(cls)
2941 self.clauses = list(clauses) if clauses else []
2942 self.group = True
2943 self.operator = operator
2944 self.group_contents = True
2945 self._is_implicitly_boolean = False
2946 return self
2948 def __iter__(self) -> Iterator[ColumnElement[Any]]:
2949 return iter(self.clauses)
2951 def __len__(self) -> int:
2952 return len(self.clauses)
2954 @property
2955 def _select_iterable(self) -> _SelectIterable:
2956 return itertools.chain.from_iterable(
2957 [elem._select_iterable for elem in self.clauses]
2958 )
2960 def append(self, clause):
2961 if self.group_contents:
2962 self.clauses.append(
2963 coercions.expect(self._text_converter_role, clause).self_group(
2964 against=self.operator
2965 )
2966 )
2967 else:
2968 self.clauses.append(
2969 coercions.expect(self._text_converter_role, clause)
2970 )
2972 @util.ro_non_memoized_property
2973 def _from_objects(self) -> List[FromClause]:
2974 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
2976 def self_group(
2977 self, against: Optional[OperatorType] = None
2978 ) -> Union[Self, Grouping[Any]]:
2979 if self.group and operators.is_precedent(self.operator, against):
2980 return Grouping(self)
2981 else:
2982 return self
2985class OperatorExpression(ColumnElement[_T]):
2986 """base for expressions that contain an operator and operands
2988 .. versionadded:: 2.0
2990 """
2992 operator: OperatorType
2993 type: TypeEngine[_T]
2995 group: bool = True
2997 @property
2998 def is_comparison(self):
2999 return operators.is_comparison(self.operator)
3001 def self_group(
3002 self, against: Optional[OperatorType] = None
3003 ) -> Union[Self, Grouping[_T]]:
3004 if (
3005 self.group
3006 and operators.is_precedent(self.operator, against)
3007 or (
3008 # a negate against a non-boolean operator
3009 # doesn't make too much sense but we should
3010 # group for that
3011 against is operators.inv
3012 and not operators.is_boolean(self.operator)
3013 )
3014 ):
3015 return Grouping(self)
3016 else:
3017 return self
3019 @property
3020 def _flattened_operator_clauses(
3021 self,
3022 ) -> typing_Tuple[ColumnElement[Any], ...]:
3023 raise NotImplementedError()
3025 @classmethod
3026 def _construct_for_op(
3027 cls,
3028 left: ColumnElement[Any],
3029 right: ColumnElement[Any],
3030 op: OperatorType,
3031 *,
3032 type_: TypeEngine[_T],
3033 negate: Optional[OperatorType] = None,
3034 modifiers: Optional[Mapping[str, Any]] = None,
3035 ) -> OperatorExpression[_T]:
3036 if operators.is_associative(op):
3037 assert (
3038 negate is None
3039 ), f"negate not supported for associative operator {op}"
3041 multi = False
3042 if getattr(
3043 left, "operator", None
3044 ) is op and type_._compare_type_affinity(left.type):
3045 multi = True
3046 left_flattened = left._flattened_operator_clauses
3047 else:
3048 left_flattened = (left,)
3050 if getattr(
3051 right, "operator", None
3052 ) is op and type_._compare_type_affinity(right.type):
3053 multi = True
3054 right_flattened = right._flattened_operator_clauses
3055 else:
3056 right_flattened = (right,)
3058 if multi:
3059 return ExpressionClauseList._construct_for_list(
3060 op,
3061 type_,
3062 *(left_flattened + right_flattened),
3063 )
3065 if right._is_collection_aggregate:
3066 negate = None
3068 return BinaryExpression(
3069 left, right, op, type_=type_, negate=negate, modifiers=modifiers
3070 )
3073class ExpressionClauseList(OperatorExpression[_T]):
3074 """Describe a list of clauses, separated by an operator,
3075 in a column expression context.
3077 :class:`.ExpressionClauseList` differs from :class:`.ClauseList` in that
3078 it represents a column-oriented DQL expression only, not an open ended
3079 list of anything comma separated.
3081 .. versionadded:: 2.0
3083 """
3085 __visit_name__ = "expression_clauselist"
3087 _traverse_internals: _TraverseInternalsType = [
3088 ("clauses", InternalTraversal.dp_clauseelement_tuple),
3089 ("operator", InternalTraversal.dp_operator),
3090 ]
3092 clauses: typing_Tuple[ColumnElement[Any], ...]
3094 group: bool
3096 def __init__(
3097 self,
3098 operator: OperatorType,
3099 *clauses: _ColumnExpressionArgument[Any],
3100 type_: Optional[_TypeEngineArgument[_T]] = None,
3101 ):
3102 self.operator = operator
3104 self.clauses = tuple(
3105 coercions.expect(
3106 roles.ExpressionElementRole, clause, apply_propagate_attrs=self
3107 )
3108 for clause in clauses
3109 )
3110 self._is_implicitly_boolean = operators.is_boolean(self.operator)
3111 self.type = type_api.to_instance(type_) # type: ignore
3113 @property
3114 def _flattened_operator_clauses(
3115 self,
3116 ) -> typing_Tuple[ColumnElement[Any], ...]:
3117 return self.clauses
3119 def __iter__(self) -> Iterator[ColumnElement[Any]]:
3120 return iter(self.clauses)
3122 def __len__(self) -> int:
3123 return len(self.clauses)
3125 @property
3126 def _select_iterable(self) -> _SelectIterable:
3127 return (self,)
3129 @util.ro_non_memoized_property
3130 def _from_objects(self) -> List[FromClause]:
3131 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
3133 def _append_inplace(self, clause: ColumnElement[Any]) -> None:
3134 self.clauses += (clause,)
3136 @classmethod
3137 def _construct_for_list(
3138 cls,
3139 operator: OperatorType,
3140 type_: TypeEngine[_T],
3141 *clauses: ColumnElement[Any],
3142 group: bool = True,
3143 ) -> ExpressionClauseList[_T]:
3144 self = cls.__new__(cls)
3145 self.group = group
3146 if group:
3147 self.clauses = tuple(
3148 c.self_group(against=operator) for c in clauses
3149 )
3150 else:
3151 self.clauses = clauses
3152 self.operator = operator
3153 self.type = type_
3154 for c in clauses:
3155 if c._propagate_attrs:
3156 self._propagate_attrs = c._propagate_attrs
3157 break
3158 return self
3160 def _negate(self) -> Any:
3161 grouped = self.self_group(against=operators.inv)
3162 assert isinstance(grouped, ColumnElement)
3163 return UnaryExpression(grouped, operator=operators.inv)
3166class BooleanClauseList(ExpressionClauseList[bool]):
3167 __visit_name__ = "expression_clauselist"
3168 inherit_cache = True
3170 def __init__(self, *arg, **kw):
3171 raise NotImplementedError(
3172 "BooleanClauseList has a private constructor"
3173 )
3175 @classmethod
3176 def _process_clauses_for_boolean(
3177 cls,
3178 operator: OperatorType,
3179 continue_on: Any,
3180 skip_on: Any,
3181 clauses: Iterable[ColumnElement[Any]],
3182 ) -> typing_Tuple[int, List[ColumnElement[Any]]]:
3183 has_continue_on = None
3185 convert_clauses = []
3187 against = operators._asbool
3188 lcc = 0
3190 for clause in clauses:
3191 if clause is continue_on:
3192 # instance of continue_on, like and_(x, y, True, z), store it
3193 # if we didn't find one already, we will use it if there
3194 # are no other expressions here.
3195 has_continue_on = clause
3196 elif clause is skip_on:
3197 # instance of skip_on, e.g. and_(x, y, False, z), cancels
3198 # the rest out
3199 convert_clauses = [clause]
3200 lcc = 1
3201 break
3202 else:
3203 if not lcc:
3204 lcc = 1
3205 else:
3206 against = operator
3207 # technically this would be len(convert_clauses) + 1
3208 # however this only needs to indicate "greater than one"
3209 lcc = 2
3210 convert_clauses.append(clause)
3212 if not convert_clauses and has_continue_on is not None:
3213 convert_clauses = [has_continue_on]
3214 lcc = 1
3216 return lcc, [c.self_group(against=against) for c in convert_clauses]
3218 @classmethod
3219 def _construct(
3220 cls,
3221 operator: OperatorType,
3222 continue_on: Any,
3223 skip_on: Any,
3224 initial_clause: Any = _NoArg.NO_ARG,
3225 *clauses: Any,
3226 **kw: Any,
3227 ) -> ColumnElement[Any]:
3228 if initial_clause is _NoArg.NO_ARG:
3229 # no elements period. deprecated use case. return an empty
3230 # ClauseList construct that generates nothing unless it has
3231 # elements added to it.
3232 name = operator.__name__
3234 util.warn_deprecated(
3235 f"Invoking {name}() without arguments is deprecated, and "
3236 f"will be disallowed in a future release. For an empty "
3237 f"""{name}() construct, use '{name}({
3238 'true()' if continue_on is True_._singleton else 'false()'
3239 }, *args)' """
3240 f"""or '{name}({
3241 'True' if continue_on is True_._singleton else 'False'
3242 }, *args)'.""",
3243 version="1.4",
3244 )
3245 return cls._construct_raw(operator)
3247 lcc, convert_clauses = cls._process_clauses_for_boolean(
3248 operator,
3249 continue_on,
3250 skip_on,
3251 [
3252 coercions.expect(roles.WhereHavingRole, clause)
3253 for clause in util.coerce_generator_arg(
3254 (initial_clause,) + clauses
3255 )
3256 ],
3257 )
3259 if lcc > 1:
3260 # multiple elements. Return regular BooleanClauseList
3261 # which will link elements against the operator.
3263 flattened_clauses = itertools.chain.from_iterable(
3264 (
3265 (c for c in to_flat._flattened_operator_clauses)
3266 if getattr(to_flat, "operator", None) is operator
3267 else (to_flat,)
3268 )
3269 for to_flat in convert_clauses
3270 )
3272 return cls._construct_raw(operator, flattened_clauses) # type: ignore # noqa: E501
3273 else:
3274 assert lcc
3275 # just one element. return it as a single boolean element,
3276 # not a list and discard the operator.
3277 return convert_clauses[0]
3279 @classmethod
3280 def _construct_for_whereclause(
3281 cls, clauses: Iterable[ColumnElement[Any]]
3282 ) -> Optional[ColumnElement[bool]]:
3283 operator, continue_on, skip_on = (
3284 operators.and_,
3285 True_._singleton,
3286 False_._singleton,
3287 )
3289 lcc, convert_clauses = cls._process_clauses_for_boolean(
3290 operator,
3291 continue_on,
3292 skip_on,
3293 clauses, # these are assumed to be coerced already
3294 )
3296 if lcc > 1:
3297 # multiple elements. Return regular BooleanClauseList
3298 # which will link elements against the operator.
3299 return cls._construct_raw(operator, convert_clauses)
3300 elif lcc == 1:
3301 # just one element. return it as a single boolean element,
3302 # not a list and discard the operator.
3303 return convert_clauses[0]
3304 else:
3305 return None
3307 @classmethod
3308 def _construct_raw(
3309 cls,
3310 operator: OperatorType,
3311 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
3312 ) -> BooleanClauseList:
3313 self = cls.__new__(cls)
3314 self.clauses = tuple(clauses) if clauses else ()
3315 self.group = True
3316 self.operator = operator
3317 self.type = type_api.BOOLEANTYPE
3318 self._is_implicitly_boolean = True
3319 return self
3321 @classmethod
3322 def and_(
3323 cls,
3324 initial_clause: Union[
3325 Literal[True], _ColumnExpressionArgument[bool], _NoArg
3326 ] = _NoArg.NO_ARG,
3327 *clauses: _ColumnExpressionArgument[bool],
3328 ) -> ColumnElement[bool]:
3329 r"""Produce a conjunction of expressions joined by ``AND``.
3331 See :func:`_sql.and_` for full documentation.
3332 """
3333 return cls._construct(
3334 operators.and_,
3335 True_._singleton,
3336 False_._singleton,
3337 initial_clause,
3338 *clauses,
3339 )
3341 @classmethod
3342 def or_(
3343 cls,
3344 initial_clause: Union[
3345 Literal[False], _ColumnExpressionArgument[bool], _NoArg
3346 ] = _NoArg.NO_ARG,
3347 *clauses: _ColumnExpressionArgument[bool],
3348 ) -> ColumnElement[bool]:
3349 """Produce a conjunction of expressions joined by ``OR``.
3351 See :func:`_sql.or_` for full documentation.
3352 """
3353 return cls._construct(
3354 operators.or_,
3355 False_._singleton,
3356 True_._singleton,
3357 initial_clause,
3358 *clauses,
3359 )
3361 @property
3362 def _select_iterable(self) -> _SelectIterable:
3363 return (self,)
3365 def self_group(
3366 self, against: Optional[OperatorType] = None
3367 ) -> Union[Self, Grouping[bool]]:
3368 if not self.clauses:
3369 return self
3370 else:
3371 return super().self_group(against=against)
3374and_ = BooleanClauseList.and_
3375or_ = BooleanClauseList.or_
3378class Tuple(ClauseList, ColumnElement[TupleAny]):
3379 """Represent a SQL tuple."""
3381 __visit_name__ = "tuple"
3383 _traverse_internals: _TraverseInternalsType = (
3384 ClauseList._traverse_internals + []
3385 )
3387 type: TupleType
3389 @util.preload_module("sqlalchemy.sql.sqltypes")
3390 def __init__(
3391 self,
3392 *clauses: _ColumnExpressionArgument[Any],
3393 types: Optional[Sequence[_TypeEngineArgument[Any]]] = None,
3394 ):
3395 sqltypes = util.preloaded.sql_sqltypes
3397 if types is None:
3398 init_clauses: List[ColumnElement[Any]] = [
3399 coercions.expect(roles.ExpressionElementRole, c)
3400 for c in clauses
3401 ]
3402 else:
3403 if len(types) != len(clauses):
3404 raise exc.ArgumentError(
3405 "Wrong number of elements for %d-tuple: %r "
3406 % (len(types), clauses)
3407 )
3408 init_clauses = [
3409 coercions.expect(
3410 roles.ExpressionElementRole,
3411 c,
3412 type_=typ if not typ._isnull else None,
3413 )
3414 for typ, c in zip(types, clauses)
3415 ]
3417 self.type = sqltypes.TupleType(*[arg.type for arg in init_clauses])
3418 super().__init__(*init_clauses)
3420 @property
3421 def _select_iterable(self) -> _SelectIterable:
3422 return (self,)
3424 def _bind_param(self, operator, obj, type_=None, expanding=False):
3425 if expanding:
3426 return BindParameter(
3427 None,
3428 value=obj,
3429 _compared_to_operator=operator,
3430 unique=True,
3431 expanding=True,
3432 type_=type_,
3433 _compared_to_type=self.type,
3434 )
3435 else:
3436 return Tuple(
3437 *[
3438 BindParameter(
3439 None,
3440 o,
3441 _compared_to_operator=operator,
3442 _compared_to_type=compared_to_type,
3443 unique=True,
3444 type_=type_,
3445 )
3446 for o, compared_to_type in zip(obj, self.type.types)
3447 ]
3448 )
3450 def self_group(self, against: Optional[OperatorType] = None) -> Self:
3451 # Tuple is parenthesized by definition.
3452 return self
3455class Case(ColumnElement[_T]):
3456 """Represent a ``CASE`` expression.
3458 :class:`.Case` is produced using the :func:`.case` factory function,
3459 as in::
3461 from sqlalchemy import case
3463 stmt = select(users_table).where(
3464 case(
3465 (users_table.c.name == "wendy", "W"),
3466 (users_table.c.name == "jack", "J"),
3467 else_="E",
3468 )
3469 )
3471 Details on :class:`.Case` usage is at :func:`.case`.
3473 .. seealso::
3475 :func:`.case`
3477 """
3479 __visit_name__ = "case"
3481 _traverse_internals: _TraverseInternalsType = [
3482 ("value", InternalTraversal.dp_clauseelement),
3483 ("whens", InternalTraversal.dp_clauseelement_tuples),
3484 ("else_", InternalTraversal.dp_clauseelement),
3485 ]
3487 # for case(), the type is derived from the whens. so for the moment
3488 # users would have to cast() the case to get a specific type
3490 whens: List[typing_Tuple[ColumnElement[bool], ColumnElement[_T]]]
3491 else_: Optional[ColumnElement[_T]]
3492 value: Optional[ColumnElement[Any]]
3494 def __init__(
3495 self,
3496 *whens: Union[
3497 typing_Tuple[_ColumnExpressionArgument[bool], Any],
3498 Mapping[Any, Any],
3499 ],
3500 value: Optional[Any] = None,
3501 else_: Optional[Any] = None,
3502 ):
3503 new_whens: Iterable[Any] = coercions._expression_collection_was_a_list(
3504 "whens", "case", whens
3505 )
3506 try:
3507 new_whens = util.dictlike_iteritems(new_whens)
3508 except TypeError:
3509 pass
3511 self.whens = [
3512 (
3513 coercions.expect(
3514 roles.ExpressionElementRole,
3515 c,
3516 apply_propagate_attrs=self,
3517 ).self_group(),
3518 coercions.expect(roles.ExpressionElementRole, r),
3519 )
3520 for (c, r) in new_whens
3521 ]
3523 if value is None:
3524 self.value = None
3525 else:
3526 self.value = coercions.expect(roles.ExpressionElementRole, value)
3528 if else_ is not None:
3529 self.else_ = coercions.expect(roles.ExpressionElementRole, else_)
3530 else:
3531 self.else_ = None
3533 type_ = next(
3534 (
3535 then.type
3536 # Iterate `whens` in reverse to match previous behaviour
3537 # where type of final element took priority
3538 for *_, then in reversed(self.whens)
3539 if not then.type._isnull
3540 ),
3541 self.else_.type if self.else_ is not None else type_api.NULLTYPE,
3542 )
3543 self.type = cast(_T, type_)
3545 @util.ro_non_memoized_property
3546 def _from_objects(self) -> List[FromClause]:
3547 return list(
3548 itertools.chain(*[x._from_objects for x in self.get_children()])
3549 )
3552class Cast(WrapsColumnExpression[_T]):
3553 """Represent a ``CAST`` expression.
3555 :class:`.Cast` is produced using the :func:`.cast` factory function,
3556 as in::
3558 from sqlalchemy import cast, Numeric
3560 stmt = select(cast(product_table.c.unit_price, Numeric(10, 4)))
3562 Details on :class:`.Cast` usage is at :func:`.cast`.
3564 .. seealso::
3566 :ref:`tutorial_casts`
3568 :func:`.cast`
3570 :func:`.try_cast`
3572 :func:`.type_coerce` - an alternative to CAST that coerces the type
3573 on the Python side only, which is often sufficient to generate the
3574 correct SQL and data coercion.
3576 """
3578 __visit_name__ = "cast"
3580 _traverse_internals: _TraverseInternalsType = [
3581 ("clause", InternalTraversal.dp_clauseelement),
3582 ("type", InternalTraversal.dp_type),
3583 ]
3585 clause: ColumnElement[Any]
3586 type: TypeEngine[_T]
3587 typeclause: TypeClause
3589 def __init__(
3590 self,
3591 expression: _ColumnExpressionArgument[Any],
3592 type_: _TypeEngineArgument[_T],
3593 ):
3594 self.type = type_api.to_instance(type_)
3595 self.clause = coercions.expect(
3596 roles.ExpressionElementRole,
3597 expression,
3598 type_=self.type,
3599 apply_propagate_attrs=self,
3600 )
3601 self.typeclause = TypeClause(self.type)
3603 @util.ro_non_memoized_property
3604 def _from_objects(self) -> List[FromClause]:
3605 return self.clause._from_objects
3607 @property
3608 def wrapped_column_expression(self):
3609 return self.clause
3612class TryCast(Cast[_T]):
3613 """Represent a TRY_CAST expression.
3615 Details on :class:`.TryCast` usage is at :func:`.try_cast`.
3617 .. seealso::
3619 :func:`.try_cast`
3621 :ref:`tutorial_casts`
3622 """
3624 __visit_name__ = "try_cast"
3625 inherit_cache = True
3628class TypeCoerce(WrapsColumnExpression[_T]):
3629 """Represent a Python-side type-coercion wrapper.
3631 :class:`.TypeCoerce` supplies the :func:`_expression.type_coerce`
3632 function; see that function for usage details.
3634 .. seealso::
3636 :func:`_expression.type_coerce`
3638 :func:`.cast`
3640 """
3642 __visit_name__ = "type_coerce"
3644 _traverse_internals: _TraverseInternalsType = [
3645 ("clause", InternalTraversal.dp_clauseelement),
3646 ("type", InternalTraversal.dp_type),
3647 ]
3649 clause: ColumnElement[Any]
3650 type: TypeEngine[_T]
3652 def __init__(
3653 self,
3654 expression: _ColumnExpressionArgument[Any],
3655 type_: _TypeEngineArgument[_T],
3656 ):
3657 self.type = type_api.to_instance(type_)
3658 self.clause = coercions.expect(
3659 roles.ExpressionElementRole,
3660 expression,
3661 type_=self.type,
3662 apply_propagate_attrs=self,
3663 )
3665 @util.ro_non_memoized_property
3666 def _from_objects(self) -> List[FromClause]:
3667 return self.clause._from_objects
3669 @HasMemoized.memoized_attribute
3670 def typed_expression(self):
3671 if isinstance(self.clause, BindParameter):
3672 bp = self.clause._clone()
3673 bp.type = self.type
3674 return bp
3675 else:
3676 return self.clause
3678 @property
3679 def wrapped_column_expression(self):
3680 return self.clause
3682 def self_group(
3683 self, against: Optional[OperatorType] = None
3684 ) -> TypeCoerce[_T]:
3685 grouped = self.clause.self_group(against=against)
3686 if grouped is not self.clause:
3687 return TypeCoerce(grouped, self.type)
3688 else:
3689 return self
3692class Extract(ColumnElement[int]):
3693 """Represent a SQL EXTRACT clause, ``extract(field FROM expr)``."""
3695 __visit_name__ = "extract"
3697 _traverse_internals: _TraverseInternalsType = [
3698 ("expr", InternalTraversal.dp_clauseelement),
3699 ("field", InternalTraversal.dp_string),
3700 ]
3702 expr: ColumnElement[Any]
3703 field: str
3705 def __init__(self, field: str, expr: _ColumnExpressionArgument[Any]):
3706 self.type = type_api.INTEGERTYPE
3707 self.field = field
3708 self.expr = coercions.expect(roles.ExpressionElementRole, expr)
3710 @util.ro_non_memoized_property
3711 def _from_objects(self) -> List[FromClause]:
3712 return self.expr._from_objects
3715class _label_reference(ColumnElement[_T]):
3716 """Wrap a column expression as it appears in a 'reference' context.
3718 This expression is any that includes an _order_by_label_element,
3719 which is a Label, or a DESC / ASC construct wrapping a Label.
3721 The production of _label_reference() should occur when an expression
3722 is added to this context; this includes the ORDER BY or GROUP BY of a
3723 SELECT statement, as well as a few other places, such as the ORDER BY
3724 within an OVER clause.
3726 """
3728 __visit_name__ = "label_reference"
3730 _traverse_internals: _TraverseInternalsType = [
3731 ("element", InternalTraversal.dp_clauseelement)
3732 ]
3734 element: ColumnElement[_T]
3736 def __init__(self, element: ColumnElement[_T]):
3737 self.element = element
3738 self._propagate_attrs = element._propagate_attrs
3740 @util.ro_non_memoized_property
3741 def _from_objects(self) -> List[FromClause]:
3742 return []
3745class _textual_label_reference(ColumnElement[Any]):
3746 __visit_name__ = "textual_label_reference"
3748 _traverse_internals: _TraverseInternalsType = [
3749 ("element", InternalTraversal.dp_string)
3750 ]
3752 def __init__(self, element: str):
3753 self.element = element
3755 @util.memoized_property
3756 def _text_clause(self) -> TextClause:
3757 return TextClause(self.element)
3760class UnaryExpression(ColumnElement[_T]):
3761 """Define a 'unary' expression.
3763 A unary expression has a single column expression
3764 and an operator. The operator can be placed on the left
3765 (where it is called the 'operator') or right (where it is called the
3766 'modifier') of the column expression.
3768 :class:`.UnaryExpression` is the basis for several unary operators
3769 including those used by :func:`.desc`, :func:`.asc`, :func:`.distinct`,
3770 :func:`.nulls_first` and :func:`.nulls_last`.
3772 """
3774 __visit_name__ = "unary"
3776 _traverse_internals: _TraverseInternalsType = [
3777 ("element", InternalTraversal.dp_clauseelement),
3778 ("operator", InternalTraversal.dp_operator),
3779 ("modifier", InternalTraversal.dp_operator),
3780 ]
3782 element: ColumnElement[Any]
3783 operator: Optional[OperatorType]
3784 modifier: Optional[OperatorType]
3786 def __init__(
3787 self,
3788 element: ColumnElement[Any],
3789 *,
3790 operator: Optional[OperatorType] = None,
3791 modifier: Optional[OperatorType] = None,
3792 type_: Optional[_TypeEngineArgument[_T]] = None,
3793 wraps_column_expression: bool = False, # legacy, not used as of 2.0.42
3794 ):
3795 self.operator = operator
3796 self.modifier = modifier
3797 self._propagate_attrs = element._propagate_attrs
3798 self.element = element.self_group(
3799 against=self.operator or self.modifier
3800 )
3802 # if type is None, we get NULLTYPE, which is our _T. But I don't
3803 # know how to get the overloads to express that correctly
3804 self.type = type_api.to_instance(type_) # type: ignore
3806 def _wraps_unnamed_column(self):
3807 ungrouped = self.element._ungroup()
3808 return (
3809 not isinstance(ungrouped, NamedColumn)
3810 or ungrouped._non_anon_label is None
3811 )
3813 @classmethod
3814 def _create_nulls_first(
3815 cls,
3816 column: _ColumnExpressionArgument[_T],
3817 ) -> UnaryExpression[_T]:
3818 return UnaryExpression(
3819 coercions.expect(roles.ByOfRole, column),
3820 modifier=operators.nulls_first_op,
3821 )
3823 @classmethod
3824 def _create_nulls_last(
3825 cls,
3826 column: _ColumnExpressionArgument[_T],
3827 ) -> UnaryExpression[_T]:
3828 return UnaryExpression(
3829 coercions.expect(roles.ByOfRole, column),
3830 modifier=operators.nulls_last_op,
3831 )
3833 @classmethod
3834 def _create_desc(
3835 cls, column: _ColumnExpressionOrStrLabelArgument[_T]
3836 ) -> UnaryExpression[_T]:
3838 return UnaryExpression(
3839 coercions.expect(roles.ByOfRole, column),
3840 modifier=operators.desc_op,
3841 )
3843 @classmethod
3844 def _create_asc(
3845 cls,
3846 column: _ColumnExpressionOrStrLabelArgument[_T],
3847 ) -> UnaryExpression[_T]:
3848 return UnaryExpression(
3849 coercions.expect(roles.ByOfRole, column),
3850 modifier=operators.asc_op,
3851 )
3853 @classmethod
3854 def _create_distinct(
3855 cls,
3856 expr: _ColumnExpressionArgument[_T],
3857 ) -> UnaryExpression[_T]:
3858 col_expr: ColumnElement[_T] = coercions.expect(
3859 roles.ExpressionElementRole, expr
3860 )
3861 return UnaryExpression(
3862 col_expr,
3863 operator=operators.distinct_op,
3864 type_=col_expr.type,
3865 )
3867 @classmethod
3868 def _create_bitwise_not(
3869 cls,
3870 expr: _ColumnExpressionArgument[_T],
3871 ) -> UnaryExpression[_T]:
3872 col_expr: ColumnElement[_T] = coercions.expect(
3873 roles.ExpressionElementRole, expr
3874 )
3875 return UnaryExpression(
3876 col_expr,
3877 operator=operators.bitwise_not_op,
3878 type_=col_expr.type,
3879 )
3881 @property
3882 def _order_by_label_element(self) -> Optional[Label[Any]]:
3883 if operators.is_order_by_modifier(self.modifier):
3884 return self.element._order_by_label_element
3885 else:
3886 return None
3888 @util.ro_non_memoized_property
3889 def _from_objects(self) -> List[FromClause]:
3890 return self.element._from_objects
3892 def _negate(self) -> ColumnElement[Any]:
3893 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
3894 return UnaryExpression(
3895 self.self_group(against=operators.inv),
3896 operator=operators.inv,
3897 type_=type_api.BOOLEANTYPE,
3898 )
3899 else:
3900 return ColumnElement._negate(self)
3902 def self_group(
3903 self, against: Optional[OperatorType] = None
3904 ) -> Union[Self, Grouping[_T]]:
3905 if self.operator and operators.is_precedent(self.operator, against):
3906 return Grouping(self)
3907 else:
3908 return self
3911class CollectionAggregate(UnaryExpression[_T]):
3912 """Forms the basis for right-hand collection operator modifiers
3913 ANY and ALL.
3915 The ANY and ALL keywords are available in different ways on different
3916 backends. On PostgreSQL, they only work for an ARRAY type. On
3917 MySQL, they only work for subqueries.
3919 """
3921 inherit_cache = True
3922 _is_collection_aggregate = True
3924 @classmethod
3925 def _create_any(
3926 cls, expr: _ColumnExpressionArgument[_T]
3927 ) -> CollectionAggregate[bool]:
3928 """create CollectionAggregate for the legacy
3929 ARRAY.Comparator.any() method"""
3930 col_expr: ColumnElement[_T] = coercions.expect(
3931 roles.ExpressionElementRole,
3932 expr,
3933 )
3934 col_expr = col_expr.self_group()
3935 return CollectionAggregate(
3936 col_expr,
3937 operator=operators.any_op,
3938 type_=type_api.BOOLEANTYPE,
3939 )
3941 @classmethod
3942 def _create_all(
3943 cls, expr: _ColumnExpressionArgument[_T]
3944 ) -> CollectionAggregate[bool]:
3945 """create CollectionAggregate for the legacy
3946 ARRAY.Comparator.all() method"""
3947 col_expr: ColumnElement[_T] = coercions.expect(
3948 roles.ExpressionElementRole,
3949 expr,
3950 )
3951 col_expr = col_expr.self_group()
3952 return CollectionAggregate(
3953 col_expr,
3954 operator=operators.all_op,
3955 type_=type_api.BOOLEANTYPE,
3956 )
3958 @util.preload_module("sqlalchemy.sql.sqltypes")
3959 def _bind_param(
3960 self,
3961 operator: operators.OperatorType,
3962 obj: Any,
3963 type_: Optional[TypeEngine[_T]] = None,
3964 expanding: bool = False,
3965 ) -> BindParameter[_T]:
3966 """For new style any_(), all_(), ensure compared literal value
3967 receives appropriate bound parameter type."""
3969 # a CollectionAggregate is specific to ARRAY or int
3970 # only. So for ARRAY case, make sure we use correct element type
3971 sqltypes = util.preloaded.sql_sqltypes
3972 if self.element.type._type_affinity is sqltypes.ARRAY:
3973 compared_to_type = cast(
3974 sqltypes.ARRAY[Any], self.element.type
3975 ).item_type
3976 else:
3977 compared_to_type = self.element.type
3979 return BindParameter(
3980 None,
3981 obj,
3982 _compared_to_operator=operator,
3983 type_=type_,
3984 _compared_to_type=compared_to_type,
3985 unique=True,
3986 expanding=expanding,
3987 )
3989 # operate and reverse_operate are hardwired to
3990 # dispatch onto the type comparator directly, so that we can
3991 # ensure "reversed" behavior.
3992 def operate(
3993 self, op: OperatorType, *other: Any, **kwargs: Any
3994 ) -> ColumnElement[_T]:
3995 if not operators.is_comparison(op):
3996 raise exc.ArgumentError(
3997 "Only comparison operators may be used with ANY/ALL"
3998 )
3999 kwargs["reverse"] = True
4000 return self.comparator.operate(operators.mirror(op), *other, **kwargs)
4002 def reverse_operate(
4003 self, op: OperatorType, other: Any, **kwargs: Any
4004 ) -> ColumnElement[_T]:
4005 # comparison operators should never call reverse_operate
4006 assert not operators.is_comparison(op)
4007 raise exc.ArgumentError(
4008 "Only comparison operators may be used with ANY/ALL"
4009 )
4012class AsBoolean(WrapsColumnExpression[bool], UnaryExpression[bool]):
4013 inherit_cache = True
4015 def __init__(self, element, operator, negate):
4016 self.element = element
4017 self.type = type_api.BOOLEANTYPE
4018 self.operator = operator
4019 self.negate = negate
4020 self.modifier = None
4021 self._is_implicitly_boolean = element._is_implicitly_boolean
4023 @property
4024 def wrapped_column_expression(self):
4025 return self.element
4027 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4028 return self
4030 def _negate(self):
4031 if isinstance(self.element, (True_, False_)):
4032 return self.element._negate()
4033 else:
4034 return AsBoolean(self.element, self.negate, self.operator)
4037class BinaryExpression(OperatorExpression[_T]):
4038 """Represent an expression that is ``LEFT <operator> RIGHT``.
4040 A :class:`.BinaryExpression` is generated automatically
4041 whenever two column expressions are used in a Python binary expression:
4043 .. sourcecode:: pycon+sql
4045 >>> from sqlalchemy.sql import column
4046 >>> column("a") + column("b")
4047 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
4048 >>> print(column("a") + column("b"))
4049 {printsql}a + b
4051 """
4053 __visit_name__ = "binary"
4055 _traverse_internals: _TraverseInternalsType = [
4056 ("left", InternalTraversal.dp_clauseelement),
4057 ("right", InternalTraversal.dp_clauseelement),
4058 ("operator", InternalTraversal.dp_operator),
4059 ("negate", InternalTraversal.dp_operator),
4060 ("modifiers", InternalTraversal.dp_plain_dict),
4061 (
4062 "type",
4063 InternalTraversal.dp_type,
4064 ),
4065 ]
4067 _cache_key_traversal = [
4068 ("left", InternalTraversal.dp_clauseelement),
4069 ("right", InternalTraversal.dp_clauseelement),
4070 ("operator", InternalTraversal.dp_operator),
4071 ("modifiers", InternalTraversal.dp_plain_dict),
4072 # "type" affects JSON CAST operators, so while redundant in most cases,
4073 # is needed for that one
4074 (
4075 "type",
4076 InternalTraversal.dp_type,
4077 ),
4078 ]
4080 _is_implicitly_boolean = True
4081 """Indicates that any database will know this is a boolean expression
4082 even if the database does not have an explicit boolean datatype.
4084 """
4086 left: ColumnElement[Any]
4087 right: ColumnElement[Any]
4088 modifiers: Mapping[str, Any]
4090 def __init__(
4091 self,
4092 left: ColumnElement[Any],
4093 right: ColumnElement[Any],
4094 operator: OperatorType,
4095 type_: Optional[_TypeEngineArgument[_T]] = None,
4096 negate: Optional[OperatorType] = None,
4097 modifiers: Optional[Mapping[str, Any]] = None,
4098 ):
4099 # allow compatibility with libraries that
4100 # refer to BinaryExpression directly and pass strings
4101 if isinstance(operator, str):
4102 operator = operators.custom_op(operator)
4103 self._orig = (left.__hash__(), right.__hash__())
4104 self._propagate_attrs = left._propagate_attrs or right._propagate_attrs
4105 self.left = left.self_group(against=operator)
4106 self.right = right.self_group(against=operator)
4107 self.operator = operator
4109 # if type is None, we get NULLTYPE, which is our _T. But I don't
4110 # know how to get the overloads to express that correctly
4111 self.type = type_api.to_instance(type_) # type: ignore
4113 self.negate = negate
4114 self._is_implicitly_boolean = operators.is_boolean(operator)
4116 if modifiers is None:
4117 self.modifiers = {}
4118 else:
4119 self.modifiers = modifiers
4121 @property
4122 def _flattened_operator_clauses(
4123 self,
4124 ) -> typing_Tuple[ColumnElement[Any], ...]:
4125 return (self.left, self.right)
4127 def __bool__(self):
4128 """Implement Python-side "bool" for BinaryExpression as a
4129 simple "identity" check for the left and right attributes,
4130 if the operator is "eq" or "ne". Otherwise the expression
4131 continues to not support "bool" like all other column expressions.
4133 The rationale here is so that ColumnElement objects can be hashable.
4134 What? Well, suppose you do this::
4136 c1, c2 = column("x"), column("y")
4137 s1 = set([c1, c2])
4139 We do that **a lot**, columns inside of sets is an extremely basic
4140 thing all over the ORM for example.
4142 So what happens if we do this? ::
4144 c1 in s1
4146 Hashing means it will normally use ``__hash__()`` of the object,
4147 but in case of hash collision, it's going to also do ``c1 == c1``
4148 and/or ``c1 == c2`` inside. Those operations need to return a
4149 True/False value. But because we override ``==`` and ``!=``, they're
4150 going to get a BinaryExpression. Hence we implement ``__bool__`` here
4151 so that these comparisons behave in this particular context mostly
4152 like regular object comparisons. Thankfully Python is OK with
4153 that! Otherwise we'd have to use special set classes for columns
4154 (which we used to do, decades ago).
4156 """
4157 if self.operator in (operators.eq, operators.ne):
4158 # this is using the eq/ne operator given int hash values,
4159 # rather than Operator, so that "bool" can be based on
4160 # identity
4161 return self.operator(*self._orig) # type: ignore
4162 else:
4163 raise TypeError("Boolean value of this clause is not defined")
4165 if typing.TYPE_CHECKING:
4167 def __invert__(
4168 self: BinaryExpression[_T],
4169 ) -> BinaryExpression[_T]: ...
4171 @util.ro_non_memoized_property
4172 def _from_objects(self) -> List[FromClause]:
4173 return self.left._from_objects + self.right._from_objects
4175 def _negate(self):
4176 if self.negate is not None:
4177 return BinaryExpression(
4178 self.left,
4179 self.right._negate_in_binary(self.negate, self.operator),
4180 self.negate,
4181 negate=self.operator,
4182 type_=self.type,
4183 modifiers=self.modifiers,
4184 )
4185 else:
4186 return self.self_group()._negate()
4189class Slice(ColumnElement[Any]):
4190 """Represent SQL for a Python array-slice object.
4192 This is not a specific SQL construct at this level, but
4193 may be interpreted by specific dialects, e.g. PostgreSQL.
4195 """
4197 __visit_name__ = "slice"
4199 _traverse_internals: _TraverseInternalsType = [
4200 ("start", InternalTraversal.dp_clauseelement),
4201 ("stop", InternalTraversal.dp_clauseelement),
4202 ("step", InternalTraversal.dp_clauseelement),
4203 ]
4205 def __init__(self, start, stop, step, _name=None):
4206 self.start = coercions.expect(
4207 roles.ExpressionElementRole,
4208 start,
4209 name=_name,
4210 type_=type_api.INTEGERTYPE,
4211 )
4212 self.stop = coercions.expect(
4213 roles.ExpressionElementRole,
4214 stop,
4215 name=_name,
4216 type_=type_api.INTEGERTYPE,
4217 )
4218 self.step = coercions.expect(
4219 roles.ExpressionElementRole,
4220 step,
4221 name=_name,
4222 type_=type_api.INTEGERTYPE,
4223 )
4224 self.type = type_api.NULLTYPE
4226 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4227 assert against is operator.getitem
4228 return self
4231class IndexExpression(BinaryExpression[Any]):
4232 """Represent the class of expressions that are like an "index"
4233 operation."""
4235 inherit_cache = True
4238class GroupedElement(DQLDMLClauseElement):
4239 """Represent any parenthesized expression"""
4241 __visit_name__ = "grouping"
4243 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4244 return self
4246 def _ungroup(self) -> ClauseElement:
4247 raise NotImplementedError()
4250class Grouping(GroupedElement, ColumnElement[_T]):
4251 """Represent a grouping within a column expression"""
4253 _traverse_internals: _TraverseInternalsType = [
4254 ("element", InternalTraversal.dp_clauseelement),
4255 ("type", InternalTraversal.dp_type),
4256 ]
4258 _cache_key_traversal = [
4259 ("element", InternalTraversal.dp_clauseelement),
4260 ]
4262 element: Union[
4263 TextClause, ClauseList, ColumnElement[_T], CompilerColumnElement
4264 ]
4266 def __init__(
4267 self,
4268 element: Union[
4269 TextClause, ClauseList, ColumnElement[_T], CompilerColumnElement
4270 ],
4271 ):
4272 self.element = element
4274 # nulltype assignment issue
4275 self.type = getattr(element, "type", type_api.NULLTYPE) # type: ignore
4276 self._propagate_attrs = element._propagate_attrs
4278 def _with_binary_element_type(self, type_):
4279 return self.__class__(self.element._with_binary_element_type(type_))
4281 def _ungroup(self) -> ColumnElement[_T]:
4282 assert isinstance(self.element, ColumnElement)
4283 return self.element._ungroup()
4285 @util.memoized_property
4286 def _is_implicitly_boolean(self):
4287 return self.element._is_implicitly_boolean
4289 @util.non_memoized_property
4290 def _tq_label(self) -> Optional[str]:
4291 return (
4292 getattr(self.element, "_tq_label", None) or self._anon_name_label
4293 )
4295 @util.non_memoized_property
4296 def _proxies(self) -> List[ColumnElement[Any]]:
4297 if isinstance(self.element, ColumnElement):
4298 return [self.element]
4299 else:
4300 return []
4302 @util.ro_non_memoized_property
4303 def _from_objects(self) -> List[FromClause]:
4304 return self.element._from_objects
4306 def __getattr__(self, attr):
4307 return getattr(self.element, attr)
4309 def __getstate__(self):
4310 return {"element": self.element, "type": self.type}
4312 def __setstate__(self, state):
4313 self.element = state["element"]
4314 self.type = state["type"]
4316 if TYPE_CHECKING:
4318 def self_group(
4319 self, against: Optional[OperatorType] = None
4320 ) -> Self: ...
4323class _OverrideBinds(Grouping[_T]):
4324 """used by cache_key->_apply_params_to_element to allow compilation /
4325 execution of a SQL element that's been cached, using an alternate set of
4326 bound parameter values.
4328 This is used by the ORM to swap new parameter values into expressions
4329 that are embedded into loader options like with_expression(),
4330 selectinload(). Previously, this task was accomplished using the
4331 .params() method which would perform a deep-copy instead. This deep
4332 copy proved to be too expensive for more complex expressions.
4334 See #11085
4336 """
4338 __visit_name__ = "override_binds"
4340 def __init__(
4341 self,
4342 element: ColumnElement[_T],
4343 bindparams: Sequence[BindParameter[Any]],
4344 replaces_params: Sequence[BindParameter[Any]],
4345 ):
4346 self.element = element
4347 self.translate = {
4348 k.key: v.value for k, v in zip(replaces_params, bindparams)
4349 }
4351 def _gen_cache_key(
4352 self, anon_map: anon_map, bindparams: List[BindParameter[Any]]
4353 ) -> Optional[typing_Tuple[Any, ...]]:
4354 """generate a cache key for the given element, substituting its bind
4355 values for the translation values present."""
4357 existing_bps: List[BindParameter[Any]] = []
4358 ck = self.element._gen_cache_key(anon_map, existing_bps)
4360 bindparams.extend(
4361 (
4362 bp._with_value(
4363 self.translate[bp.key], maintain_key=True, required=False
4364 )
4365 if bp.key in self.translate
4366 else bp
4367 )
4368 for bp in existing_bps
4369 )
4371 return ck
4374class Over(ColumnElement[_T]):
4375 """Represent an OVER clause.
4377 This is a special operator against a so-called
4378 "window" function, as well as any aggregate function,
4379 which produces results relative to the result set
4380 itself. Most modern SQL backends now support window functions.
4382 """
4384 __visit_name__ = "over"
4386 _traverse_internals: _TraverseInternalsType = [
4387 ("element", InternalTraversal.dp_clauseelement),
4388 ("order_by", InternalTraversal.dp_clauseelement),
4389 ("partition_by", InternalTraversal.dp_clauseelement),
4390 ("range_", InternalTraversal.dp_clauseelement),
4391 ("rows", InternalTraversal.dp_clauseelement),
4392 ("groups", InternalTraversal.dp_clauseelement),
4393 ]
4395 order_by: Optional[ClauseList] = None
4396 partition_by: Optional[ClauseList] = None
4398 element: ColumnElement[_T]
4399 """The underlying expression object to which this :class:`.Over`
4400 object refers."""
4402 range_: Optional[_FrameClause]
4403 rows: Optional[_FrameClause]
4404 groups: Optional[_FrameClause]
4406 def __init__(
4407 self,
4408 element: ColumnElement[_T],
4409 partition_by: Optional[_ByArgument] = None,
4410 order_by: Optional[_ByArgument] = None,
4411 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4412 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4413 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4414 ):
4415 self.element = element
4416 if order_by is not None:
4417 self.order_by = ClauseList(
4418 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4419 )
4420 if partition_by is not None:
4421 self.partition_by = ClauseList(
4422 *util.to_list(partition_by),
4423 _literal_as_text_role=roles.ByOfRole,
4424 )
4426 if sum(bool(item) for item in (range_, rows, groups)) > 1:
4427 raise exc.ArgumentError(
4428 "only one of 'rows', 'range_', or 'groups' may be provided"
4429 )
4430 else:
4431 self.range_ = _FrameClause(range_) if range_ else None
4432 self.rows = _FrameClause(rows) if rows else None
4433 self.groups = _FrameClause(groups) if groups else None
4435 if not TYPE_CHECKING:
4437 @util.memoized_property
4438 def type(self) -> TypeEngine[_T]: # noqa: A001
4439 return self.element.type
4441 @util.ro_non_memoized_property
4442 def _from_objects(self) -> List[FromClause]:
4443 return list(
4444 itertools.chain(
4445 *[
4446 c._from_objects
4447 for c in (self.element, self.partition_by, self.order_by)
4448 if c is not None
4449 ]
4450 )
4451 )
4454class _FrameClauseType(Enum):
4455 RANGE_UNBOUNDED = 0
4456 RANGE_CURRENT = 1
4457 RANGE_PRECEDING = 2
4458 RANGE_FOLLOWING = 3
4461class _FrameClause(ClauseElement):
4462 """indicate the 'rows' or 'range' field of a window function, e.g. using
4463 :class:`.Over`.
4465 .. versionadded:: 2.1
4467 """
4469 __visit_name__ = "frame_clause"
4471 _traverse_internals: _TraverseInternalsType = [
4472 ("lower_integer_bind", InternalTraversal.dp_clauseelement),
4473 ("upper_integer_bind", InternalTraversal.dp_clauseelement),
4474 ("lower_type", InternalTraversal.dp_plain_obj),
4475 ("upper_type", InternalTraversal.dp_plain_obj),
4476 ]
4478 def __init__(
4479 self,
4480 range_: typing_Tuple[Optional[int], Optional[int]],
4481 ):
4482 try:
4483 r0, r1 = range_
4484 except (ValueError, TypeError) as ve:
4485 raise exc.ArgumentError("2-tuple expected for range/rows") from ve
4487 if r0 is None:
4488 self.lower_type = _FrameClauseType.RANGE_UNBOUNDED
4489 self.lower_integer_bind = None
4490 else:
4491 try:
4492 lower_integer = int(r0)
4493 except ValueError as err:
4494 raise exc.ArgumentError(
4495 "Integer or None expected for range value"
4496 ) from err
4497 else:
4498 if lower_integer == 0:
4499 self.lower_type = _FrameClauseType.RANGE_CURRENT
4500 self.lower_integer_bind = None
4501 elif lower_integer < 0:
4502 self.lower_type = _FrameClauseType.RANGE_PRECEDING
4503 self.lower_integer_bind = literal(
4504 abs(lower_integer), type_api.INTEGERTYPE
4505 )
4506 else:
4507 self.lower_type = _FrameClauseType.RANGE_FOLLOWING
4508 self.lower_integer_bind = literal(
4509 lower_integer, type_api.INTEGERTYPE
4510 )
4512 if r1 is None:
4513 self.upper_type = _FrameClauseType.RANGE_UNBOUNDED
4514 self.upper_integer_bind = None
4515 else:
4516 try:
4517 upper_integer = int(r1)
4518 except ValueError as err:
4519 raise exc.ArgumentError(
4520 "Integer or None expected for range value"
4521 ) from err
4522 else:
4523 if upper_integer == 0:
4524 self.upper_type = _FrameClauseType.RANGE_CURRENT
4525 self.upper_integer_bind = None
4526 elif upper_integer < 0:
4527 self.upper_type = _FrameClauseType.RANGE_PRECEDING
4528 self.upper_integer_bind = literal(
4529 abs(upper_integer), type_api.INTEGERTYPE
4530 )
4531 else:
4532 self.upper_type = _FrameClauseType.RANGE_FOLLOWING
4533 self.upper_integer_bind = literal(
4534 upper_integer, type_api.INTEGERTYPE
4535 )
4538class AggregateOrderBy(WrapsColumnExpression[_T]):
4539 """Represent an aggregate ORDER BY expression.
4541 This is a special operator against aggregate functions such as
4542 ``array_agg()``, ``json_arrayagg()`` ``string_agg()``, etc. that provides
4543 for an ORDER BY expression, using a syntax that's compatible with
4544 the backend.
4546 :class:`.AggregateOrderBy` is a generalized version of the
4547 :class:`.WithinGroup` construct, the latter of which always provides a
4548 "WITHIN GROUP (ORDER BY ...)" expression. :class:`.AggregateOrderBy` will
4549 also compile to "WITHIN GROUP (ORDER BY ...)" on backends such as Oracle
4550 and SQL Server that don't have another style of aggregate function
4551 ordering.
4553 .. versionadded:: 2.1
4556 """
4558 __visit_name__ = "aggregateorderby"
4560 _traverse_internals: _TraverseInternalsType = [
4561 ("element", InternalTraversal.dp_clauseelement),
4562 ("order_by", InternalTraversal.dp_clauseelement),
4563 ]
4565 order_by: ClauseList
4567 def __init__(
4568 self,
4569 element: Union[FunctionElement[_T], FunctionFilter[_T]],
4570 *order_by: _ColumnExpressionArgument[Any],
4571 ):
4572 self.element = element
4573 if not order_by:
4574 raise TypeError("at least one ORDER BY element is required")
4575 self.order_by = ClauseList(
4576 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4577 )
4579 if not TYPE_CHECKING:
4581 @util.memoized_property
4582 def type(self) -> TypeEngine[_T]: # noqa: A001
4583 return self.element.type
4585 @property
4586 def wrapped_column_expression(self) -> ColumnElement[_T]:
4587 return self.element
4589 def __reduce__(self):
4590 return self.__class__, (self.element,) + (
4591 tuple(self.order_by) if self.order_by is not None else ()
4592 )
4594 def over(
4595 self,
4596 *,
4597 partition_by: Optional[_ByArgument] = None,
4598 order_by: Optional[_ByArgument] = None,
4599 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4600 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4601 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4602 ) -> Over[_T]:
4603 """Produce an OVER clause against this :class:`.WithinGroup`
4604 construct.
4606 This function has the same signature as that of
4607 :meth:`.FunctionElement.over`.
4609 """
4610 return Over(
4611 self,
4612 partition_by=partition_by,
4613 order_by=order_by,
4614 range_=range_,
4615 rows=rows,
4616 groups=groups,
4617 )
4619 @overload
4620 def filter(self) -> Self: ...
4622 @overload
4623 def filter(
4624 self,
4625 __criterion0: _ColumnExpressionArgument[bool],
4626 *criterion: _ColumnExpressionArgument[bool],
4627 ) -> FunctionFilter[_T]: ...
4629 def filter(
4630 self, *criterion: _ColumnExpressionArgument[bool]
4631 ) -> Union[Self, FunctionFilter[_T]]:
4632 """Produce a FILTER clause against this function."""
4633 if not criterion:
4634 return self
4635 return FunctionFilter(self, *criterion)
4637 @util.ro_non_memoized_property
4638 def _from_objects(self) -> List[FromClause]:
4639 return list(
4640 itertools.chain(
4641 *[
4642 c._from_objects
4643 for c in (self.element, self.order_by)
4644 if c is not None
4645 ]
4646 )
4647 )
4650class WithinGroup(AggregateOrderBy[_T]):
4651 """Represent a WITHIN GROUP (ORDER BY) clause.
4653 This is a special operator against so-called
4654 "ordered set aggregate" and "hypothetical
4655 set aggregate" functions, including ``percentile_cont()``,
4656 ``rank()``, ``dense_rank()``, etc.
4658 It's supported only by certain database backends, such as PostgreSQL,
4659 Oracle Database and MS SQL Server.
4661 The :class:`.WithinGroup` construct extracts its type from the
4662 method :meth:`.FunctionElement.within_group_type`. If this returns
4663 ``None``, the function's ``.type`` is used.
4665 """
4667 __visit_name__ = "withingroup"
4668 inherit_cache = True
4670 if not TYPE_CHECKING:
4672 @util.memoized_property
4673 def type(self) -> TypeEngine[_T]: # noqa: A001
4674 wgt = self.element.within_group_type(self)
4675 if wgt is not None:
4676 return wgt
4677 else:
4678 return self.element.type
4681class FunctionFilter(Generative, ColumnElement[_T]):
4682 """Represent a function FILTER clause.
4684 This is a special operator against aggregate and window functions,
4685 which controls which rows are passed to it.
4686 It's supported only by certain database backends.
4688 Invocation of :class:`.FunctionFilter` is via
4689 :meth:`.FunctionElement.filter`::
4691 func.count(1).filter(True)
4693 .. seealso::
4695 :meth:`.FunctionElement.filter`
4697 """
4699 __visit_name__ = "funcfilter"
4701 _traverse_internals: _TraverseInternalsType = [
4702 ("func", InternalTraversal.dp_clauseelement),
4703 ("criterion", InternalTraversal.dp_clauseelement),
4704 ]
4706 criterion: Optional[ColumnElement[bool]] = None
4708 def __init__(
4709 self,
4710 func: Union[FunctionElement[_T], AggregateOrderBy[_T]],
4711 *criterion: _ColumnExpressionArgument[bool],
4712 ):
4713 self.func = func
4714 self.filter.non_generative(self, *criterion) # type: ignore
4716 @_generative
4717 def filter(self, *criterion: _ColumnExpressionArgument[bool]) -> Self:
4718 """Produce an additional FILTER against the function.
4720 This method adds additional criteria to the initial criteria
4721 set up by :meth:`.FunctionElement.filter`.
4723 Multiple criteria are joined together at SQL render time
4724 via ``AND``.
4727 """
4729 for crit in list(criterion):
4730 crit = coercions.expect(roles.WhereHavingRole, crit)
4732 if self.criterion is not None:
4733 self.criterion = self.criterion & crit
4734 else:
4735 self.criterion = crit
4737 return self
4739 def over(
4740 self,
4741 partition_by: Optional[
4742 Union[
4743 Iterable[_ColumnExpressionArgument[Any]],
4744 _ColumnExpressionArgument[Any],
4745 ]
4746 ] = None,
4747 order_by: Optional[
4748 Union[
4749 Iterable[_ColumnExpressionArgument[Any]],
4750 _ColumnExpressionArgument[Any],
4751 ]
4752 ] = None,
4753 range_: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4754 rows: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4755 groups: Optional[typing_Tuple[Optional[int], Optional[int]]] = None,
4756 ) -> Over[_T]:
4757 """Produce an OVER clause against this filtered function.
4759 Used against aggregate or so-called "window" functions,
4760 for database backends that support window functions.
4762 The expression::
4764 func.rank().filter(MyClass.y > 5).over(order_by="x")
4766 is shorthand for::
4768 from sqlalchemy import over, funcfilter
4770 over(funcfilter(func.rank(), MyClass.y > 5), order_by="x")
4772 See :func:`_expression.over` for a full description.
4774 """
4775 return Over(
4776 self,
4777 partition_by=partition_by,
4778 order_by=order_by,
4779 range_=range_,
4780 rows=rows,
4781 groups=groups,
4782 )
4784 def within_group(
4785 self, *order_by: _ColumnExpressionArgument[Any]
4786 ) -> WithinGroup[_T]:
4787 """Produce a WITHIN GROUP (ORDER BY expr) clause against
4788 this function.
4789 """
4790 return WithinGroup(self, *order_by)
4792 def within_group_type(
4793 self, within_group: WithinGroup[_T]
4794 ) -> Optional[TypeEngine[_T]]:
4795 return None
4797 def self_group(
4798 self, against: Optional[OperatorType] = None
4799 ) -> Union[Self, Grouping[_T]]:
4800 if operators.is_precedent(operators.filter_op, against):
4801 return Grouping(self)
4802 else:
4803 return self
4805 if not TYPE_CHECKING:
4807 @util.memoized_property
4808 def type(self) -> TypeEngine[_T]: # noqa: A001
4809 return self.func.type
4811 @util.ro_non_memoized_property
4812 def _from_objects(self) -> List[FromClause]:
4813 return list(
4814 itertools.chain(
4815 *[
4816 c._from_objects
4817 for c in (self.func, self.criterion)
4818 if c is not None
4819 ]
4820 )
4821 )
4824class NamedColumn(KeyedColumnElement[_T]):
4825 is_literal = False
4826 table: Optional[FromClause] = None
4827 name: str
4828 key: str
4830 def _compare_name_for_result(self, other):
4831 return (hasattr(other, "name") and self.name == other.name) or (
4832 hasattr(other, "_label") and self._label == other._label
4833 )
4835 @util.ro_memoized_property
4836 def description(self) -> str:
4837 return self.name
4839 @HasMemoized.memoized_attribute
4840 def _tq_key_label(self) -> Optional[str]:
4841 """table qualified label based on column key.
4843 for table-bound columns this is <tablename>_<column key/proxy key>;
4845 all other expressions it resolves to key/proxy key.
4847 """
4848 proxy_key = self._proxy_key
4849 if proxy_key and proxy_key != self.name:
4850 return self._gen_tq_label(proxy_key)
4851 else:
4852 return self._tq_label
4854 @HasMemoized.memoized_attribute
4855 def _tq_label(self) -> Optional[str]:
4856 """table qualified label based on column name.
4858 for table-bound columns this is <tablename>_<columnname>; all other
4859 expressions it resolves to .name.
4861 """
4862 return self._gen_tq_label(self.name)
4864 @HasMemoized.memoized_attribute
4865 def _render_label_in_columns_clause(self):
4866 return True
4868 @HasMemoized.memoized_attribute
4869 def _non_anon_label(self):
4870 return self.name
4872 def _gen_tq_label(
4873 self, name: str, dedupe_on_key: bool = True
4874 ) -> Optional[str]:
4875 return name
4877 def _bind_param(
4878 self,
4879 operator: OperatorType,
4880 obj: Any,
4881 type_: Optional[TypeEngine[_T]] = None,
4882 expanding: bool = False,
4883 ) -> BindParameter[_T]:
4884 return BindParameter(
4885 self.key,
4886 obj,
4887 _compared_to_operator=operator,
4888 _compared_to_type=self.type,
4889 type_=type_,
4890 unique=True,
4891 expanding=expanding,
4892 )
4894 def _make_proxy(
4895 self,
4896 selectable: FromClause,
4897 *,
4898 primary_key: ColumnSet,
4899 foreign_keys: Set[KeyedColumnElement[Any]],
4900 name: Optional[str] = None,
4901 key: Optional[str] = None,
4902 name_is_truncatable: bool = False,
4903 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
4904 disallow_is_literal: bool = False,
4905 **kw: Any,
4906 ) -> typing_Tuple[str, ColumnClause[_T]]:
4907 c = ColumnClause(
4908 (
4909 coercions.expect(roles.TruncatedLabelRole, name or self.name)
4910 if name_is_truncatable
4911 else (name or self.name)
4912 ),
4913 type_=self.type,
4914 _selectable=selectable,
4915 is_literal=False,
4916 )
4918 c._propagate_attrs = selectable._propagate_attrs
4919 if name is None:
4920 c.key = self.key
4921 if compound_select_cols:
4922 c._proxies = list(compound_select_cols)
4923 else:
4924 c._proxies = [self]
4926 if selectable._is_clone_of is not None:
4927 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
4928 return c.key, c
4931_PS = ParamSpec("_PS")
4934class Label(roles.LabeledColumnExprRole[_T], NamedColumn[_T]):
4935 """Represents a column label (AS).
4937 Represent a label, as typically applied to any column-level
4938 element using the ``AS`` sql keyword.
4940 """
4942 __visit_name__ = "label"
4944 _traverse_internals: _TraverseInternalsType = [
4945 ("name", InternalTraversal.dp_anon_name),
4946 ("type", InternalTraversal.dp_type),
4947 ("_element", InternalTraversal.dp_clauseelement),
4948 ]
4950 _cache_key_traversal = [
4951 ("name", InternalTraversal.dp_anon_name),
4952 ("_element", InternalTraversal.dp_clauseelement),
4953 ]
4955 _element: ColumnElement[_T]
4956 name: str
4958 def __init__(
4959 self,
4960 name: Optional[str],
4961 element: _ColumnExpressionArgument[_T],
4962 type_: Optional[_TypeEngineArgument[_T]] = None,
4963 ):
4964 orig_element = element
4965 element = coercions.expect(
4966 roles.ExpressionElementRole,
4967 element,
4968 apply_propagate_attrs=self,
4969 )
4970 while isinstance(element, Label):
4971 # TODO: this is only covered in test_text.py, but nothing
4972 # fails if it's removed. determine rationale
4973 element = element.element
4975 if name:
4976 self.name = name
4977 else:
4978 self.name = _anonymous_label.safe_construct(
4979 id(self), getattr(element, "name", "anon")
4980 )
4981 if isinstance(orig_element, Label):
4982 # TODO: no coverage for this block, again would be in
4983 # test_text.py where the resolve_label concept is important
4984 self._resolve_label = orig_element._label
4986 self.key = self._tq_label = self._tq_key_label = self.name
4987 self._element = element
4989 self.type = (
4990 type_api.to_instance(type_)
4991 if type_ is not None
4992 else self._element.type
4993 )
4995 self._proxies = [element]
4997 def __reduce__(self):
4998 return self.__class__, (self.name, self._element, self.type)
5000 @HasMemoized.memoized_attribute
5001 def _render_label_in_columns_clause(self):
5002 return True
5004 def _bind_param(self, operator, obj, type_=None, expanding=False):
5005 return BindParameter(
5006 None,
5007 obj,
5008 _compared_to_operator=operator,
5009 type_=type_,
5010 _compared_to_type=self.type,
5011 unique=True,
5012 expanding=expanding,
5013 )
5015 @util.memoized_property
5016 def _is_implicitly_boolean(self):
5017 return self.element._is_implicitly_boolean
5019 @HasMemoized.memoized_attribute
5020 def _allow_label_resolve(self):
5021 return self.element._allow_label_resolve
5023 @property
5024 def _order_by_label_element(self):
5025 return self
5027 def as_reference(self) -> _label_reference[_T]:
5028 """refer to this labeled expression in a clause such as GROUP BY,
5029 ORDER BY etc. as the label name itself, without expanding
5030 into the full expression.
5032 .. versionadded:: 2.1
5034 """
5035 return _label_reference(self)
5037 @HasMemoized.memoized_attribute
5038 def element(self) -> ColumnElement[_T]:
5039 return self._element.self_group(against=operators.as_)
5041 def self_group(self, against: Optional[OperatorType] = None) -> Label[_T]:
5042 return self._apply_to_inner(self._element.self_group, against=against)
5044 def _negate(self):
5045 return self._apply_to_inner(self._element._negate)
5047 def _apply_to_inner(
5048 self,
5049 fn: Callable[_PS, ColumnElement[_T]],
5050 *arg: _PS.args,
5051 **kw: _PS.kwargs,
5052 ) -> Label[_T]:
5053 sub_element = fn(*arg, **kw)
5054 if sub_element is not self._element:
5055 return Label(self.name, sub_element, type_=self.type)
5056 else:
5057 return self
5059 @property
5060 def primary_key(self): # type: ignore[override]
5061 return self.element.primary_key
5063 @property
5064 def foreign_keys(self): # type: ignore[override]
5065 return self.element.foreign_keys
5067 def _copy_internals(
5068 self,
5069 *,
5070 clone: _CloneCallableType = _clone,
5071 anonymize_labels: bool = False,
5072 **kw: Any,
5073 ) -> None:
5074 self._reset_memoizations()
5075 self._element = clone(self._element, **kw)
5076 if anonymize_labels:
5077 self.name = _anonymous_label.safe_construct(
5078 id(self), getattr(self.element, "name", "anon")
5079 )
5080 self.key = self._tq_label = self._tq_key_label = self.name
5082 @util.ro_non_memoized_property
5083 def _from_objects(self) -> List[FromClause]:
5084 return self.element._from_objects
5086 def _make_proxy(
5087 self,
5088 selectable: FromClause,
5089 *,
5090 primary_key: ColumnSet,
5091 foreign_keys: Set[KeyedColumnElement[Any]],
5092 name: Optional[str] = None,
5093 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5094 **kw: Any,
5095 ) -> typing_Tuple[str, ColumnClause[_T]]:
5096 name = self.name if not name else name
5098 key, e = self.element._make_proxy(
5099 selectable,
5100 name=name,
5101 disallow_is_literal=True,
5102 name_is_truncatable=isinstance(name, _truncated_label),
5103 compound_select_cols=compound_select_cols,
5104 primary_key=primary_key,
5105 foreign_keys=foreign_keys,
5106 )
5108 # there was a note here to remove this assertion, which was here
5109 # to determine if we later could support a use case where
5110 # the key and name of a label are separate. But I don't know what
5111 # that case was. For now, this is an unexpected case that occurs
5112 # when a label name conflicts with other columns and select()
5113 # is attempting to disambiguate an explicit label, which is not what
5114 # the user would want. See issue #6090.
5115 if key != self.name and not isinstance(self.name, _anonymous_label):
5116 raise exc.InvalidRequestError(
5117 "Label name %s is being renamed to an anonymous label due "
5118 "to disambiguation "
5119 "which is not supported right now. Please use unique names "
5120 "for explicit labels." % (self.name)
5121 )
5123 e._propagate_attrs = selectable._propagate_attrs
5124 e._proxies.append(self)
5125 if self.type is not None:
5126 e.type = self.type
5128 return self.key, e
5131class ColumnClause(
5132 roles.DDLReferredColumnRole,
5133 roles.LabeledColumnExprRole[_T],
5134 roles.StrAsPlainColumnRole,
5135 Immutable,
5136 NamedColumn[_T],
5137):
5138 """Represents a column expression from any textual string.
5140 The :class:`.ColumnClause`, a lightweight analogue to the
5141 :class:`_schema.Column` class, is typically invoked using the
5142 :func:`_expression.column` function, as in::
5144 from sqlalchemy import column
5146 id, name = column("id"), column("name")
5147 stmt = select(id, name).select_from("user")
5149 The above statement would produce SQL like:
5151 .. sourcecode:: sql
5153 SELECT id, name FROM user
5155 :class:`.ColumnClause` is the immediate superclass of the schema-specific
5156 :class:`_schema.Column` object. While the :class:`_schema.Column`
5157 class has all the
5158 same capabilities as :class:`.ColumnClause`, the :class:`.ColumnClause`
5159 class is usable by itself in those cases where behavioral requirements
5160 are limited to simple SQL expression generation. The object has none of
5161 the associations with schema-level metadata or with execution-time
5162 behavior that :class:`_schema.Column` does,
5163 so in that sense is a "lightweight"
5164 version of :class:`_schema.Column`.
5166 Full details on :class:`.ColumnClause` usage is at
5167 :func:`_expression.column`.
5169 .. seealso::
5171 :func:`_expression.column`
5173 :class:`_schema.Column`
5175 """
5177 table: Optional[FromClause]
5178 is_literal: bool
5180 __visit_name__ = "column"
5182 _traverse_internals: _TraverseInternalsType = [
5183 ("name", InternalTraversal.dp_anon_name),
5184 ("type", InternalTraversal.dp_type),
5185 ("table", InternalTraversal.dp_clauseelement),
5186 ("is_literal", InternalTraversal.dp_boolean),
5187 ]
5189 onupdate: Optional[DefaultGenerator] = None
5190 default: Optional[DefaultGenerator] = None
5191 server_default: Optional[FetchedValue] = None
5192 server_onupdate: Optional[FetchedValue] = None
5194 _is_multiparam_column = False
5196 @property
5197 def _is_star(self): # type: ignore[override]
5198 return self.is_literal and self.name == "*"
5200 def __init__(
5201 self,
5202 text: str,
5203 type_: Optional[_TypeEngineArgument[_T]] = None,
5204 is_literal: bool = False,
5205 _selectable: Optional[FromClause] = None,
5206 ):
5207 self.key = self.name = text
5208 self.table = _selectable
5210 # if type is None, we get NULLTYPE, which is our _T. But I don't
5211 # know how to get the overloads to express that correctly
5212 self.type = type_api.to_instance(type_) # type: ignore
5214 self.is_literal = is_literal
5216 def get_children(self, *, column_tables=False, **kw):
5217 # override base get_children() to not return the Table
5218 # or selectable that is parent to this column. Traversals
5219 # expect the columns of tables and subqueries to be leaf nodes.
5220 return []
5222 @property
5223 def entity_namespace(self):
5224 if self.table is not None:
5225 return self.table.entity_namespace
5226 else:
5227 return super().entity_namespace
5229 def _clone(self, detect_subquery_cols=False, **kw):
5230 if (
5231 detect_subquery_cols
5232 and self.table is not None
5233 and self.table._is_subquery
5234 ):
5235 clone = kw.pop("clone")
5236 table = clone(self.table, **kw)
5237 new = table.c.corresponding_column(self)
5238 return new
5240 return super()._clone(**kw)
5242 @HasMemoized_ro_memoized_attribute
5243 def _from_objects(self) -> List[FromClause]:
5244 t = self.table
5245 if t is not None:
5246 return [t]
5247 else:
5248 return []
5250 @HasMemoized.memoized_attribute
5251 def _render_label_in_columns_clause(self):
5252 return self.table is not None
5254 @property
5255 def _ddl_label(self):
5256 return self._gen_tq_label(self.name, dedupe_on_key=False)
5258 def _compare_name_for_result(self, other):
5259 if (
5260 self.is_literal
5261 or self.table is None
5262 or self.table._is_textual
5263 or not hasattr(other, "proxy_set")
5264 or (
5265 isinstance(other, ColumnClause)
5266 and (
5267 other.is_literal
5268 or other.table is None
5269 or other.table._is_textual
5270 )
5271 )
5272 ):
5273 return (hasattr(other, "name") and self.name == other.name) or (
5274 hasattr(other, "_tq_label")
5275 and self._tq_label == other._tq_label
5276 )
5277 else:
5278 return other.proxy_set.intersection(self.proxy_set)
5280 def _gen_tq_label(
5281 self, name: str, dedupe_on_key: bool = True
5282 ) -> Optional[str]:
5283 """generate table-qualified label
5285 for a table-bound column this is <tablename>_<columnname>.
5287 used primarily for LABEL_STYLE_TABLENAME_PLUS_COL
5288 as well as the .columns collection on a Join object.
5290 """
5291 label: str
5292 t = self.table
5293 if self.is_literal:
5294 return None
5295 elif t is not None and is_named_from_clause(t):
5296 if has_schema_attr(t) and t.schema:
5297 label = (
5298 t.schema.replace(".", "_") + "_" + t.name + ("_" + name)
5299 )
5300 else:
5301 assert not TYPE_CHECKING or isinstance(t, NamedFromClause)
5302 label = t.name + ("_" + name)
5304 # propagate name quoting rules for labels.
5305 if is_quoted_name(name) and name.quote is not None:
5306 if is_quoted_name(label):
5307 label.quote = name.quote
5308 else:
5309 label = quoted_name(label, name.quote)
5310 elif is_quoted_name(t.name) and t.name.quote is not None:
5311 # can't get this situation to occur, so let's
5312 # assert false on it for now
5313 assert not isinstance(label, quoted_name)
5314 label = quoted_name(label, t.name.quote)
5316 if dedupe_on_key:
5317 # ensure the label name doesn't conflict with that of an
5318 # existing column. note that this implies that any Column
5319 # must **not** set up its _label before its parent table has
5320 # all of its other Column objects set up. There are several
5321 # tables in the test suite which will fail otherwise; example:
5322 # table "owner" has columns "name" and "owner_name". Therefore
5323 # column owner.name cannot use the label "owner_name", it has
5324 # to be "owner_name_1".
5325 if label in t.c:
5326 _label = label
5327 counter = 1
5328 while _label in t.c:
5329 _label = label + f"_{counter}"
5330 counter += 1
5331 label = _label
5333 return coercions.expect(roles.TruncatedLabelRole, label)
5335 else:
5336 return name
5338 def _make_proxy(
5339 self,
5340 selectable: FromClause,
5341 *,
5342 primary_key: ColumnSet,
5343 foreign_keys: Set[KeyedColumnElement[Any]],
5344 name: Optional[str] = None,
5345 key: Optional[str] = None,
5346 name_is_truncatable: bool = False,
5347 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5348 disallow_is_literal: bool = False,
5349 **kw: Any,
5350 ) -> typing_Tuple[str, ColumnClause[_T]]:
5351 # the "is_literal" flag normally should never be propagated; a proxied
5352 # column is always a SQL identifier and never the actual expression
5353 # being evaluated. however, there is a case where the "is_literal" flag
5354 # might be used to allow the given identifier to have a fixed quoting
5355 # pattern already, so maintain the flag for the proxy unless a
5356 # :class:`.Label` object is creating the proxy. See [ticket:4730].
5357 is_literal = (
5358 not disallow_is_literal
5359 and self.is_literal
5360 and (
5361 # note this does not accommodate for quoted_name differences
5362 # right now
5363 name is None
5364 or name == self.name
5365 )
5366 )
5367 c = self._constructor(
5368 (
5369 coercions.expect(roles.TruncatedLabelRole, name or self.name)
5370 if name_is_truncatable
5371 else (name or self.name)
5372 ),
5373 type_=self.type,
5374 _selectable=selectable,
5375 is_literal=is_literal,
5376 )
5377 c._propagate_attrs = selectable._propagate_attrs
5378 if name is None:
5379 c.key = self.key
5380 if compound_select_cols:
5381 c._proxies = list(compound_select_cols)
5382 else:
5383 c._proxies = [self]
5385 if selectable._is_clone_of is not None:
5386 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5387 return c.key, c
5390class TableValuedColumn(NamedColumn[_T]):
5391 __visit_name__ = "table_valued_column"
5393 _traverse_internals: _TraverseInternalsType = [
5394 ("name", InternalTraversal.dp_anon_name),
5395 ("type", InternalTraversal.dp_type),
5396 ("scalar_alias", InternalTraversal.dp_clauseelement),
5397 ]
5399 def __init__(self, scalar_alias: NamedFromClause, type_: TypeEngine[_T]):
5400 self.scalar_alias = scalar_alias
5401 self.key = self.name = scalar_alias.name
5402 self.type = type_
5404 def _copy_internals(
5405 self, clone: _CloneCallableType = _clone, **kw: Any
5406 ) -> None:
5407 self.scalar_alias = clone(self.scalar_alias, **kw)
5408 self.key = self.name = self.scalar_alias.name
5410 @util.ro_non_memoized_property
5411 def _from_objects(self) -> List[FromClause]:
5412 return [self.scalar_alias]
5415class CollationClause(ColumnElement[str]):
5416 __visit_name__ = "collation"
5418 _traverse_internals: _TraverseInternalsType = [
5419 ("collation", InternalTraversal.dp_string)
5420 ]
5422 @classmethod
5423 @util.preload_module("sqlalchemy.sql.sqltypes")
5424 def _create_collation_expression(
5425 cls, expression: _ColumnExpressionArgument[str], collation: str
5426 ) -> BinaryExpression[str]:
5428 sqltypes = util.preloaded.sql_sqltypes
5430 expr = coercions.expect(roles.ExpressionElementRole[str], expression)
5432 if expr.type._type_affinity is sqltypes.String:
5433 collate_type = expr.type._with_collation(collation)
5434 else:
5435 collate_type = expr.type
5437 return BinaryExpression(
5438 expr,
5439 CollationClause(collation),
5440 operators.collate,
5441 type_=collate_type,
5442 )
5444 def __init__(self, collation):
5445 self.collation = collation
5448class _IdentifiedClause(Executable, ClauseElement):
5449 __visit_name__ = "identified"
5451 def __init__(self, ident):
5452 self.ident = ident
5455class SavepointClause(_IdentifiedClause):
5456 __visit_name__ = "savepoint"
5457 inherit_cache = False
5460class RollbackToSavepointClause(_IdentifiedClause):
5461 __visit_name__ = "rollback_to_savepoint"
5462 inherit_cache = False
5465class ReleaseSavepointClause(_IdentifiedClause):
5466 __visit_name__ = "release_savepoint"
5467 inherit_cache = False
5470class quoted_name(util.MemoizedSlots, str):
5471 """Represent a SQL identifier combined with quoting preferences.
5473 :class:`.quoted_name` is a Python unicode/str subclass which
5474 represents a particular identifier name along with a
5475 ``quote`` flag. This ``quote`` flag, when set to
5476 ``True`` or ``False``, overrides automatic quoting behavior
5477 for this identifier in order to either unconditionally quote
5478 or to not quote the name. If left at its default of ``None``,
5479 quoting behavior is applied to the identifier on a per-backend basis
5480 based on an examination of the token itself.
5482 A :class:`.quoted_name` object with ``quote=True`` is also
5483 prevented from being modified in the case of a so-called
5484 "name normalize" option. Certain database backends, such as
5485 Oracle Database, Firebird, and DB2 "normalize" case-insensitive names
5486 as uppercase. The SQLAlchemy dialects for these backends
5487 convert from SQLAlchemy's lower-case-means-insensitive convention
5488 to the upper-case-means-insensitive conventions of those backends.
5489 The ``quote=True`` flag here will prevent this conversion from occurring
5490 to support an identifier that's quoted as all lower case against
5491 such a backend.
5493 The :class:`.quoted_name` object is normally created automatically
5494 when specifying the name for key schema constructs such as
5495 :class:`_schema.Table`, :class:`_schema.Column`, and others.
5496 The class can also be
5497 passed explicitly as the name to any function that receives a name which
5498 can be quoted. Such as to use the :meth:`_engine.Engine.has_table`
5499 method with
5500 an unconditionally quoted name::
5502 from sqlalchemy import create_engine
5503 from sqlalchemy import inspect
5504 from sqlalchemy.sql import quoted_name
5506 engine = create_engine("oracle+oracledb://some_dsn")
5507 print(inspect(engine).has_table(quoted_name("some_table", True)))
5509 The above logic will run the "has table" logic against the Oracle Database
5510 backend, passing the name exactly as ``"some_table"`` without converting to
5511 upper case.
5513 """
5515 __slots__ = "quote", "lower", "upper"
5517 quote: Optional[bool]
5519 @overload
5520 @classmethod
5521 def construct(cls, value: str, quote: Optional[bool]) -> quoted_name: ...
5523 @overload
5524 @classmethod
5525 def construct(cls, value: None, quote: Optional[bool]) -> None: ...
5527 @classmethod
5528 def construct(
5529 cls, value: Optional[str], quote: Optional[bool]
5530 ) -> Optional[quoted_name]:
5531 if value is None:
5532 return None
5533 else:
5534 return quoted_name(value, quote)
5536 def __new__(cls, value: str, quote: Optional[bool]) -> quoted_name:
5537 assert (
5538 value is not None
5539 ), "use quoted_name.construct() for None passthrough"
5540 if isinstance(value, cls) and (quote is None or value.quote == quote):
5541 return value
5542 self = super().__new__(cls, value)
5544 self.quote = quote
5545 return self
5547 def __reduce__(self):
5548 return quoted_name, (str(self), self.quote)
5550 def _memoized_method_lower(self):
5551 if self.quote:
5552 return self
5553 else:
5554 return str(self).lower()
5556 def _memoized_method_upper(self):
5557 if self.quote:
5558 return self
5559 else:
5560 return str(self).upper()
5563def _find_columns(clause: ClauseElement) -> Set[ColumnClause[Any]]:
5564 """locate Column objects within the given expression."""
5566 cols: Set[ColumnClause[Any]] = set()
5567 traverse(clause, {}, {"column": cols.add})
5568 return cols
5571def _type_from_args(args: Sequence[ColumnElement[_T]]) -> TypeEngine[_T]:
5572 for a in args:
5573 if not a.type._isnull:
5574 return a.type
5575 else:
5576 return type_api.NULLTYPE # type: ignore
5579def _corresponding_column_or_error(fromclause, column, require_embedded=False):
5580 c = fromclause.corresponding_column(
5581 column, require_embedded=require_embedded
5582 )
5583 if c is None:
5584 raise exc.InvalidRequestError(
5585 "Given column '%s', attached to table '%s', "
5586 "failed to locate a corresponding column from table '%s'"
5587 % (column, getattr(column, "table", None), fromclause.description)
5588 )
5589 return c
5592class _memoized_property_but_not_nulltype(
5593 util.memoized_property["TypeEngine[_T]"]
5594):
5595 """memoized property, but dont memoize NullType"""
5597 def __get__(self, obj, cls):
5598 if obj is None:
5599 return self
5600 result = self.fget(obj)
5601 if not result._isnull:
5602 obj.__dict__[self.__name__] = result
5603 return result
5606class AnnotatedColumnElement(Annotated):
5607 _Annotated__element: ColumnElement[Any]
5609 def __init__(self, element, values):
5610 Annotated.__init__(self, element, values)
5611 for attr in (
5612 "comparator",
5613 "_proxy_key",
5614 "_tq_key_label",
5615 "_tq_label",
5616 "_non_anon_label",
5617 "type",
5618 ):
5619 self.__dict__.pop(attr, None)
5620 for attr in ("name", "key", "table"):
5621 if self.__dict__.get(attr, False) is None:
5622 self.__dict__.pop(attr)
5624 def _with_annotations(self, values):
5625 clone = super()._with_annotations(values)
5626 for attr in (
5627 "comparator",
5628 "_proxy_key",
5629 "_tq_key_label",
5630 "_tq_label",
5631 "_non_anon_label",
5632 ):
5633 clone.__dict__.pop(attr, None)
5634 return clone
5636 @util.memoized_property
5637 def name(self):
5638 """pull 'name' from parent, if not present"""
5639 return self._Annotated__element.name
5641 @_memoized_property_but_not_nulltype
5642 def type(self):
5643 """pull 'type' from parent and don't cache if null.
5645 type is routinely changed on existing columns within the
5646 mapped_column() initialization process, and "type" is also consulted
5647 during the creation of SQL expressions. Therefore it can change after
5648 it was already retrieved. At the same time we don't want annotated
5649 objects having overhead when expressions are produced, so continue
5650 to memoize, but only when we have a non-null type.
5652 """
5653 return self._Annotated__element.type
5655 @util.memoized_property
5656 def table(self):
5657 """pull 'table' from parent, if not present"""
5658 return self._Annotated__element.table
5660 @util.memoized_property
5661 def key(self):
5662 """pull 'key' from parent, if not present"""
5663 return self._Annotated__element.key
5665 @util.memoized_property
5666 def info(self) -> _InfoType:
5667 if TYPE_CHECKING:
5668 assert isinstance(self._Annotated__element, Column)
5669 return self._Annotated__element.info
5671 @util.memoized_property
5672 def _anon_name_label(self) -> str:
5673 return self._Annotated__element._anon_name_label
5676class _truncated_label(quoted_name):
5677 """A unicode subclass used to identify symbolic "
5678 "names that may require truncation."""
5680 __slots__ = ()
5682 def __new__(cls, value: str, quote: Optional[bool] = None) -> Any:
5683 quote = getattr(value, "quote", quote)
5684 # return super(_truncated_label, cls).__new__(cls, value, quote, True)
5685 return super().__new__(cls, value, quote)
5687 def __reduce__(self) -> Any:
5688 return self.__class__, (str(self), self.quote)
5690 def apply_map(self, map_: Mapping[str, Any]) -> str:
5691 return self
5694class conv(_truncated_label):
5695 """Mark a string indicating that a name has already been converted
5696 by a naming convention.
5698 This is a string subclass that indicates a name that should not be
5699 subject to any further naming conventions.
5701 E.g. when we create a :class:`.Constraint` using a naming convention
5702 as follows::
5704 m = MetaData(
5705 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5706 )
5707 t = Table(
5708 "t", m, Column("x", Integer), CheckConstraint("x > 5", name="x5")
5709 )
5711 The name of the above constraint will be rendered as ``"ck_t_x5"``.
5712 That is, the existing name ``x5`` is used in the naming convention as the
5713 ``constraint_name`` token.
5715 In some situations, such as in migration scripts, we may be rendering
5716 the above :class:`.CheckConstraint` with a name that's already been
5717 converted. In order to make sure the name isn't double-modified, the
5718 new name is applied using the :func:`_schema.conv` marker. We can
5719 use this explicitly as follows::
5722 m = MetaData(
5723 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5724 )
5725 t = Table(
5726 "t",
5727 m,
5728 Column("x", Integer),
5729 CheckConstraint("x > 5", name=conv("ck_t_x5")),
5730 )
5732 Where above, the :func:`_schema.conv` marker indicates that the constraint
5733 name here is final, and the name will render as ``"ck_t_x5"`` and not
5734 ``"ck_t_ck_t_x5"``
5736 .. seealso::
5738 :ref:`constraint_naming_conventions`
5740 """
5742 __slots__ = ()
5745# for backwards compatibility in case
5746# someone is re-implementing the
5747# _truncated_identifier() sequence in a custom
5748# compiler
5749_generated_label = _truncated_label
5750_anonymous_label_escape = re.compile(r"[%\(\) \$]+")
5753class _anonymous_label(_truncated_label):
5754 """A unicode subclass used to identify anonymously
5755 generated names."""
5757 __slots__ = ()
5759 @classmethod
5760 def safe_construct_with_key(
5761 cls, seed: int, body: str, sanitize_key: bool = False
5762 ) -> typing_Tuple[_anonymous_label, str]:
5763 # need to escape chars that interfere with format
5764 # strings in any case, issue #8724
5765 body = _anonymous_label_escape.sub("_", body)
5767 if sanitize_key:
5768 # sanitize_key is then an extra step used by BindParameter
5769 body = body.strip("_")
5771 key = f"{seed} {body.replace('%', '%%')}"
5772 label = _anonymous_label(f"%({key})s")
5773 return label, key
5775 @classmethod
5776 def safe_construct(
5777 cls, seed: int, body: str, sanitize_key: bool = False
5778 ) -> _anonymous_label:
5779 # need to escape chars that interfere with format
5780 # strings in any case, issue #8724
5781 body = _anonymous_label_escape.sub("_", body)
5783 if sanitize_key:
5784 # sanitize_key is then an extra step used by BindParameter
5785 body = body.strip("_")
5787 return _anonymous_label(f"%({seed} {body.replace('%', '%%')})s")
5789 def __add__(self, other: str) -> _anonymous_label:
5790 if "%" in other and not isinstance(other, _anonymous_label):
5791 other = str(other).replace("%", "%%")
5792 else:
5793 other = str(other)
5795 return _anonymous_label(
5796 quoted_name(
5797 str.__add__(self, other),
5798 self.quote,
5799 )
5800 )
5802 def __radd__(self, other: str) -> _anonymous_label:
5803 if "%" in other and not isinstance(other, _anonymous_label):
5804 other = str(other).replace("%", "%%")
5805 else:
5806 other = str(other)
5808 return _anonymous_label(
5809 quoted_name(
5810 str.__add__(other, self),
5811 self.quote,
5812 )
5813 )
5815 def apply_map(self, map_: Mapping[str, Any]) -> str:
5816 if self.quote is not None:
5817 # preserve quoting only if necessary
5818 return quoted_name(self % map_, self.quote)
5819 else:
5820 # else skip the constructor call
5821 return self % map_