Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# sql/elements.py
2# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7# mypy: allow-untyped-defs, allow-untyped-calls
9"""Core SQL expression elements, including :class:`_expression.ClauseElement`,
10:class:`_expression.ColumnElement`, and derived classes.
12"""
14from __future__ import annotations
16from decimal import Decimal
17from enum import Enum
18import itertools
19import operator
20import re
21import typing
22from typing import AbstractSet
23from typing import Any
24from typing import Callable
25from typing import cast
26from typing import Dict
27from typing import FrozenSet
28from typing import Generic
29from typing import Iterable
30from typing import Iterator
31from typing import List
32from typing import Literal
33from typing import Mapping
34from typing import Optional
35from typing import overload
36from typing import ParamSpec
37from typing import Sequence
38from typing import Set
39from typing import Tuple as typing_Tuple
40from typing import Type
41from typing import TYPE_CHECKING
42from typing import TypeVar
43from typing import Union
45from . import coercions
46from . import operators
47from . import roles
48from . import traversals
49from . import type_api
50from ._typing import has_schema_attr
51from ._typing import is_named_from_clause
52from ._typing import is_quoted_name
53from ._typing import is_tuple_type
54from .annotation import Annotated
55from .annotation import SupportsWrappingAnnotations
56from .base import _clone
57from .base import _expand_cloned
58from .base import _generative
59from .base import _NoArg
60from .base import Executable
61from .base import ExecutableStatement
62from .base import Generative
63from .base import HasMemoized
64from .base import Immutable
65from .base import NO_ARG
66from .base import SingletonConstant
67from .cache_key import MemoizedHasCacheKey
68from .cache_key import NO_CACHE
69from .coercions import _document_text_coercion # noqa
70from .operators import ColumnOperators
71from .operators import OperatorClass
72from .traversals import HasCopyInternals
73from .visitors import cloned_traverse
74from .visitors import ExternallyTraversible
75from .visitors import InternalTraversal
76from .visitors import traverse
77from .visitors import Visitable
78from .. import exc
79from .. import inspection
80from .. import util
81from ..util import deprecated
82from ..util import HasMemoized_ro_memoized_attribute
83from ..util import TypingOnly
84from ..util.compat import Template
85from ..util.typing import Self
86from ..util.typing import TupleAny
87from ..util.typing import Unpack
90if typing.TYPE_CHECKING:
91 from ._typing import _ByArgument
92 from ._typing import _ColumnExpressionArgument
93 from ._typing import _ColumnExpressionOrStrLabelArgument
94 from ._typing import _HasDialect
95 from ._typing import _InfoType
96 from ._typing import _OnlyColumnArgument
97 from ._typing import _PropagateAttrsType
98 from ._typing import _TypeEngineArgument
99 from .base import _EntityNamespace
100 from .base import ColumnSet
101 from .cache_key import _CacheKeyTraversalType
102 from .cache_key import CacheKey
103 from .compiler import Compiled
104 from .compiler import SQLCompiler
105 from .functions import FunctionElement
106 from .operators import OperatorType
107 from .schema import Column
108 from .schema import DefaultGenerator
109 from .schema import FetchedValue
110 from .schema import ForeignKey
111 from .selectable import _SelectIterable
112 from .selectable import FromClause
113 from .selectable import NamedFromClause
114 from .selectable import TextualSelect
115 from .sqltypes import TupleType
116 from .type_api import TypeEngine
117 from .visitors import _CloneCallableType
118 from .visitors import _TraverseInternalsType
119 from .visitors import anon_map
120 from ..engine import Connection
121 from ..engine import Dialect
122 from ..engine.interfaces import _CoreMultiExecuteParams
123 from ..engine.interfaces import _CoreSingleExecuteParams
124 from ..engine.interfaces import CacheStats
125 from ..engine.interfaces import CompiledCacheType
126 from ..engine.interfaces import CoreExecuteOptionsParameter
127 from ..engine.interfaces import SchemaTranslateMapType
128 from ..engine.result import Result
131_NUMERIC = Union[float, Decimal]
132_NUMBER = Union[float, int, Decimal]
134_T = TypeVar("_T", bound="Any")
135_T_co = TypeVar("_T_co", bound=Any, covariant=True)
136_OPT = TypeVar("_OPT", bound="Any")
137_NT = TypeVar("_NT", bound="_NUMERIC")
139_NMT = TypeVar("_NMT", bound="_NUMBER")
142@overload
143def literal(
144 value: Any,
145 type_: _TypeEngineArgument[_T],
146 literal_execute: bool = False,
147) -> BindParameter[_T]: ...
150@overload
151def literal(
152 value: _T,
153 type_: None = None,
154 literal_execute: bool = False,
155) -> BindParameter[_T]: ...
158@overload
159def literal(
160 value: Any,
161 type_: Optional[_TypeEngineArgument[Any]] = None,
162 literal_execute: bool = False,
163) -> BindParameter[Any]: ...
166def literal(
167 value: Any,
168 type_: Optional[_TypeEngineArgument[Any]] = None,
169 literal_execute: bool = False,
170) -> BindParameter[Any]:
171 r"""Return a literal clause, bound to a bind parameter.
173 Literal clauses are created automatically when non-
174 :class:`_expression.ClauseElement` objects (such as strings, ints, dates,
175 etc.) are
176 used in a comparison operation with a :class:`_expression.ColumnElement`
177 subclass,
178 such as a :class:`~sqlalchemy.schema.Column` object. Use this function
179 to force the generation of a literal clause, which will be created as a
180 :class:`BindParameter` with a bound value.
182 :param value: the value to be bound. Can be any Python object supported by
183 the underlying DB-API, or is translatable via the given type argument.
185 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine` which will
186 provide bind-parameter translation for this literal.
188 :param literal_execute: optional bool, when True, the SQL engine will
189 attempt to render the bound value directly in the SQL statement at
190 execution time rather than providing as a parameter value.
192 .. versionadded:: 2.0
194 """
195 return coercions.expect(
196 roles.LiteralValueRole,
197 value,
198 type_=type_,
199 literal_execute=literal_execute,
200 )
203def literal_column(
204 text: str, type_: Optional[_TypeEngineArgument[_T]] = None
205) -> ColumnClause[_T]:
206 r"""Produce a :class:`.ColumnClause` object that has the
207 :paramref:`_expression.column.is_literal` flag set to True.
209 :func:`_expression.literal_column` is similar to
210 :func:`_expression.column`, except that
211 it is more often used as a "standalone" column expression that renders
212 exactly as stated; while :func:`_expression.column`
213 stores a string name that
214 will be assumed to be part of a table and may be quoted as such,
215 :func:`_expression.literal_column` can be that,
216 or any other arbitrary column-oriented
217 expression.
219 :param text: the text of the expression; can be any SQL expression.
220 Quoting rules will not be applied. To specify a column-name expression
221 which should be subject to quoting rules, use the :func:`column`
222 function.
224 :param type\_: an optional :class:`~sqlalchemy.types.TypeEngine`
225 object which will
226 provide result-set translation and additional expression semantics for
227 this column. If left as ``None`` the type will be :class:`.NullType`.
229 .. seealso::
231 :func:`_expression.column`
233 :func:`_expression.text`
235 :ref:`tutorial_select_arbitrary_text`
237 """
238 return ColumnClause(text, type_=type_, is_literal=True)
241class CompilerElement(Visitable):
242 """base class for SQL elements that can be compiled to produce a
243 SQL string.
245 .. versionadded:: 2.0
247 """
249 __slots__ = ()
250 __visit_name__ = "compiler_element"
252 supports_execution = False
254 stringify_dialect = "default"
256 @util.preload_module("sqlalchemy.engine.default")
257 @util.preload_module("sqlalchemy.engine.url")
258 def compile(
259 self,
260 bind: Optional[_HasDialect] = None,
261 dialect: Optional[Dialect] = None,
262 **kw: Any,
263 ) -> Compiled:
264 """Compile this SQL expression.
266 The return value is a :class:`~.Compiled` object.
267 Calling ``str()`` or ``unicode()`` on the returned value will yield a
268 string representation of the result. The
269 :class:`~.Compiled` object also can return a
270 dictionary of bind parameter names and values
271 using the ``params`` accessor.
273 :param bind: An :class:`.Connection` or :class:`.Engine` which
274 can provide a :class:`.Dialect` in order to generate a
275 :class:`.Compiled` object. If the ``bind`` and
276 ``dialect`` parameters are both omitted, a default SQL compiler
277 is used.
279 :param column_keys: Used for INSERT and UPDATE statements, a list of
280 column names which should be present in the VALUES clause of the
281 compiled statement. If ``None``, all columns from the target table
282 object are rendered.
284 :param dialect: A :class:`.Dialect` instance which can generate
285 a :class:`.Compiled` object. This argument takes precedence over
286 the ``bind`` argument.
288 :param compile_kwargs: optional dictionary of additional parameters
289 that will be passed through to the compiler within all "visit"
290 methods. This allows any custom flag to be passed through to
291 a custom compilation construct, for example. It is also used
292 for the case of passing the ``literal_binds`` flag through::
294 from sqlalchemy.sql import table, column, select
296 t = table("t", column("x"))
298 s = select(t).where(t.c.x == 5)
300 print(s.compile(compile_kwargs={"literal_binds": True}))
302 .. seealso::
304 :ref:`faq_sql_expression_string`
306 """
308 if dialect is None:
309 if bind:
310 dialect = bind.dialect
311 elif self.stringify_dialect == "default":
312 dialect = self._default_dialect()
313 else:
314 url = util.preloaded.engine_url
315 dialect = url.URL.create(
316 self.stringify_dialect
317 ).get_dialect()()
319 return self._compiler(dialect, **kw)
321 def _default_dialect(self):
322 default = util.preloaded.engine_default
323 return default.StrCompileDialect()
325 def _compiler(self, dialect: Dialect, **kw: Any) -> Compiled:
326 """Return a compiler appropriate for this ClauseElement, given a
327 Dialect."""
329 if TYPE_CHECKING:
330 assert isinstance(self, ClauseElement)
331 return dialect.statement_compiler(dialect, self, **kw)
333 def __str__(self) -> str:
334 return str(self.compile())
337@inspection._self_inspects
338class ClauseElement(
339 SupportsWrappingAnnotations,
340 MemoizedHasCacheKey,
341 HasCopyInternals,
342 ExternallyTraversible,
343 CompilerElement,
344):
345 """Base class for elements of a programmatically constructed SQL
346 expression.
348 """
350 __visit_name__ = "clause"
352 if TYPE_CHECKING:
354 @util.memoized_property
355 def _propagate_attrs(self) -> _PropagateAttrsType:
356 """like annotations, however these propagate outwards liberally
357 as SQL constructs are built, and are set up at construction time.
359 """
360 ...
362 else:
363 _propagate_attrs = util.EMPTY_DICT
365 @util.ro_memoized_property
366 def description(self) -> Optional[str]:
367 return None
369 _is_clone_of: Optional[Self] = None
371 is_clause_element = True
372 is_selectable = False
373 is_dml = False
374 _is_column_element = False
375 _is_keyed_column_element = False
376 _is_table = False
377 _gen_static_annotations_cache_key = False
378 _is_textual = False
379 _is_from_clause = False
380 _is_returns_rows = False
381 _is_text_clause = False
382 _is_from_container = False
383 _is_select_container = False
384 _is_select_base = False
385 _is_select_statement = False
386 _is_bind_parameter = False
387 _is_clause_list = False
388 _is_lambda_element = False
389 _is_singleton_constant = False
390 _is_immutable = False
391 _is_star = False
393 @property
394 def _order_by_label_element(self) -> Optional[Label[Any]]:
395 return None
397 _cache_key_traversal: _CacheKeyTraversalType = None
399 negation_clause: ColumnElement[bool]
401 if typing.TYPE_CHECKING:
403 def get_children(
404 self, *, omit_attrs: typing_Tuple[str, ...] = ..., **kw: Any
405 ) -> Iterable[ClauseElement]: ...
407 @util.ro_non_memoized_property
408 def _from_objects(self) -> List[FromClause]:
409 return []
411 def _set_propagate_attrs(self, values: Mapping[str, Any]) -> Self:
412 # usually, self._propagate_attrs is empty here. one case where it's
413 # not is a subquery against ORM select, that is then pulled as a
414 # property of an aliased class. should all be good
416 # assert not self._propagate_attrs
418 self._propagate_attrs = util.immutabledict(values)
419 return self
421 def _default_compiler(self) -> SQLCompiler:
422 dialect = self._default_dialect()
423 return dialect.statement_compiler(dialect, self) # type: ignore
425 def _clone(self, **kw: Any) -> Self:
426 """Create a shallow copy of this ClauseElement.
428 This method may be used by a generative API. Its also used as
429 part of the "deep" copy afforded by a traversal that combines
430 the _copy_internals() method.
432 """
434 skip = self._memoized_keys
435 c = self.__class__.__new__(self.__class__)
437 if skip:
438 # ensure this iteration remains atomic
439 c.__dict__ = {
440 k: v for k, v in self.__dict__.copy().items() if k not in skip
441 }
442 else:
443 c.__dict__ = self.__dict__.copy()
445 # this is a marker that helps to "equate" clauses to each other
446 # when a Select returns its list of FROM clauses. the cloning
447 # process leaves around a lot of remnants of the previous clause
448 # typically in the form of column expressions still attached to the
449 # old table.
450 cc = self._is_clone_of
451 c._is_clone_of = cc if cc is not None else self
452 return c
454 def _negate_in_binary(self, negated_op, original_op):
455 """a hook to allow the right side of a binary expression to respond
456 to a negation of the binary expression.
458 Used for the special case of expanding bind parameter with IN.
460 """
461 return self
463 def _with_binary_element_type(self, type_):
464 """in the context of binary expression, convert the type of this
465 object to the one given.
467 applies only to :class:`_expression.ColumnElement` classes.
469 """
470 return self
472 @property
473 def _constructor(self): # type: ignore[override]
474 """return the 'constructor' for this ClauseElement.
476 This is for the purposes for creating a new object of
477 this type. Usually, its just the element's __class__.
478 However, the "Annotated" version of the object overrides
479 to return the class of its proxied element.
481 """
482 return self.__class__
484 @HasMemoized.memoized_attribute
485 def _cloned_set(self):
486 """Return the set consisting all cloned ancestors of this
487 ClauseElement.
489 Includes this ClauseElement. This accessor tends to be used for
490 FromClause objects to identify 'equivalent' FROM clauses, regardless
491 of transformative operations.
493 """
494 s = util.column_set()
495 f: Optional[ClauseElement] = self
497 # note this creates a cycle, asserted in test_memusage. however,
498 # turning this into a plain @property adds tends of thousands of method
499 # calls to Core / ORM performance tests, so the small overhead
500 # introduced by the relatively small amount of short term cycles
501 # produced here is preferable
502 while f is not None:
503 s.add(f)
504 f = f._is_clone_of
505 return s
507 def _de_clone(self):
508 while self._is_clone_of is not None:
509 self = self._is_clone_of
510 return self
512 @util.ro_non_memoized_property
513 def entity_namespace(self) -> _EntityNamespace:
514 raise AttributeError(
515 "This SQL expression has no entity namespace "
516 "with which to filter from."
517 )
519 def __getstate__(self):
520 d = self.__dict__.copy()
521 d.pop("_is_clone_of", None)
522 d.pop("_generate_cache_key", None)
523 return d
525 def _execute_on_connection(
526 self,
527 connection: Connection,
528 distilled_params: _CoreMultiExecuteParams,
529 execution_options: CoreExecuteOptionsParameter,
530 ) -> Result[Unpack[TupleAny]]:
531 if self.supports_execution:
532 if TYPE_CHECKING:
533 assert isinstance(self, Executable)
534 return connection._execute_clauseelement(
535 self, distilled_params, execution_options
536 )
537 else:
538 raise exc.ObjectNotExecutableError(self)
540 def _execute_on_scalar(
541 self,
542 connection: Connection,
543 distilled_params: _CoreMultiExecuteParams,
544 execution_options: CoreExecuteOptionsParameter,
545 ) -> Any:
546 """an additional hook for subclasses to provide a different
547 implementation for connection.scalar() vs. connection.execute().
549 .. versionadded:: 2.0
551 """
552 return self._execute_on_connection(
553 connection, distilled_params, execution_options
554 ).scalar()
556 def _get_embedded_bindparams(self) -> Sequence[BindParameter[Any]]:
557 """Return the list of :class:`.BindParameter` objects embedded in the
558 object.
560 This accomplishes the same purpose as ``visitors.traverse()`` or
561 similar would provide, however by making use of the cache key
562 it takes advantage of memoization of the key to result in fewer
563 net method calls, assuming the statement is also going to be
564 executed.
566 """
568 key = self._generate_cache_key()
569 if key is None:
570 bindparams: List[BindParameter[Any]] = []
572 traverse(self, {}, {"bindparam": bindparams.append})
573 return bindparams
575 else:
576 return key.bindparams
578 def unique_params(
579 self,
580 __optionaldict: Optional[Dict[str, Any]] = None,
581 /,
582 **kwargs: Any,
583 ) -> Self:
584 """Return a copy with :func:`_expression.bindparam` elements
585 replaced.
587 Same functionality as :meth:`_expression.ClauseElement.params`,
588 except adds `unique=True`
589 to affected bind parameters so that multiple statements can be
590 used.
592 """
593 return self._replace_params(True, __optionaldict, kwargs)
595 def params(
596 self,
597 __optionaldict: Optional[Mapping[str, Any]] = None,
598 /,
599 **kwargs: Any,
600 ) -> Self:
601 """Return a copy with :func:`_expression.bindparam` elements
602 replaced.
604 Returns a copy of this ClauseElement with
605 :func:`_expression.bindparam`
606 elements replaced with values taken from the given dictionary::
608 >>> clause = column("x") + bindparam("foo")
609 >>> print(clause.compile().params)
610 {'foo':None}
611 >>> print(clause.params({"foo": 7}).compile().params)
612 {'foo':7}
614 """
615 return self._replace_params(False, __optionaldict, kwargs)
617 @deprecated(
618 "2.1",
619 "The params() and unique_params() methods on non-statement "
620 "ClauseElement objects are deprecated; params() is now limited to "
621 "statement level objects such as select(), insert(), union(), etc. ",
622 )
623 def _replace_params(
624 self,
625 unique: bool,
626 optionaldict: Optional[Mapping[str, Any]],
627 kwargs: Dict[str, Any],
628 ) -> Self:
629 if optionaldict:
630 kwargs.update(optionaldict)
632 def visit_bindparam(bind: BindParameter[Any]) -> None:
633 if bind.key in kwargs:
634 bind.value = kwargs[bind.key]
635 bind.required = False
636 if unique:
637 bind._convert_to_unique()
639 return cloned_traverse(
640 self,
641 {"maintain_key": True, "detect_subquery_cols": True},
642 {"bindparam": visit_bindparam},
643 )
645 def compare(self, other: ClauseElement, **kw: Any) -> bool:
646 r"""Compare this :class:`_expression.ClauseElement` to
647 the given :class:`_expression.ClauseElement`.
649 Subclasses should override the default behavior, which is a
650 straight identity comparison.
652 \**kw are arguments consumed by subclass ``compare()`` methods and
653 may be used to modify the criteria for comparison
654 (see :class:`_expression.ColumnElement`).
656 """
657 return traversals.compare(self, other, **kw)
659 def self_group(
660 self, against: Optional[OperatorType] = None
661 ) -> ClauseElement:
662 """Apply a 'grouping' to this :class:`_expression.ClauseElement`.
664 This method is overridden by subclasses to return a "grouping"
665 construct, i.e. parenthesis. In particular it's used by "binary"
666 expressions to provide a grouping around themselves when placed into a
667 larger expression, as well as by :func:`_expression.select`
668 constructs when placed into the FROM clause of another
669 :func:`_expression.select`. (Note that subqueries should be
670 normally created using the :meth:`_expression.Select.alias` method,
671 as many
672 platforms require nested SELECT statements to be named).
674 As expressions are composed together, the application of
675 :meth:`self_group` is automatic - end-user code should never
676 need to use this method directly. Note that SQLAlchemy's
677 clause constructs take operator precedence into account -
678 so parenthesis might not be needed, for example, in
679 an expression like ``x OR (y AND z)`` - AND takes precedence
680 over OR.
682 The base :meth:`self_group` method of
683 :class:`_expression.ClauseElement`
684 just returns self.
685 """
686 return self
688 def _ungroup(self) -> ClauseElement:
689 """Return this :class:`_expression.ClauseElement`
690 without any groupings.
691 """
693 return self
695 def _compile_w_cache(
696 self,
697 dialect: Dialect,
698 *,
699 compiled_cache: Optional[CompiledCacheType],
700 column_keys: List[str],
701 for_executemany: bool = False,
702 schema_translate_map: Optional[SchemaTranslateMapType] = None,
703 **kw: Any,
704 ) -> tuple[
705 Compiled,
706 Sequence[BindParameter[Any]] | None,
707 _CoreSingleExecuteParams | None,
708 CacheStats,
709 ]:
710 elem_cache_key: Optional[CacheKey]
712 if compiled_cache is not None and dialect._supports_statement_cache:
713 elem_cache_key = self._generate_cache_key()
714 else:
715 elem_cache_key = None
717 extracted_params: Optional[Sequence[BindParameter[Any]]]
718 if elem_cache_key is not None:
719 if TYPE_CHECKING:
720 assert compiled_cache is not None
722 cache_key, extracted_params, param_dict = elem_cache_key
723 key = (
724 dialect,
725 cache_key,
726 tuple(column_keys),
727 bool(schema_translate_map),
728 for_executemany,
729 )
730 compiled_sql = compiled_cache.get(key)
732 if compiled_sql is None:
733 cache_hit = dialect.CACHE_MISS
734 compiled_sql = self._compiler(
735 dialect,
736 cache_key=elem_cache_key,
737 column_keys=column_keys,
738 for_executemany=for_executemany,
739 schema_translate_map=schema_translate_map,
740 **kw,
741 )
742 # ensure that params of the current statement are not
743 # left in the cache
744 assert not compiled_sql._collect_params # type: ignore[attr-defined] # noqa: E501
745 compiled_cache[key] = compiled_sql
746 else:
747 cache_hit = dialect.CACHE_HIT
748 else:
749 param_dict = None
750 extracted_params = None
751 compiled_sql = self._compiler(
752 dialect,
753 cache_key=None,
754 column_keys=column_keys,
755 for_executemany=for_executemany,
756 schema_translate_map=schema_translate_map,
757 **kw,
758 )
759 # here instead the params need to be extracted, since we don't
760 # have them otherwise
761 assert compiled_sql._collect_params # type: ignore[attr-defined] # noqa: E501
763 if not dialect._supports_statement_cache:
764 cache_hit = dialect.NO_DIALECT_SUPPORT
765 elif compiled_cache is None:
766 cache_hit = dialect.CACHING_DISABLED
767 else:
768 cache_hit = dialect.NO_CACHE_KEY
770 return compiled_sql, extracted_params, param_dict, cache_hit
772 def __invert__(self):
773 # undocumented element currently used by the ORM for
774 # relationship.contains()
775 if hasattr(self, "negation_clause"):
776 return self.negation_clause
777 else:
778 return self._negate()
780 def _negate(self) -> ClauseElement:
781 # TODO: this code is uncovered and in all likelihood is not included
782 # in any codepath. So this should raise NotImplementedError in 2.1
783 grouped = self.self_group(against=operators.inv)
784 assert isinstance(grouped, ColumnElement)
785 return UnaryExpression(grouped, operator=operators.inv)
787 def __bool__(self):
788 raise TypeError("Boolean value of this clause is not defined")
790 def __repr__(self):
791 friendly = self.description
792 if friendly is None:
793 return object.__repr__(self)
794 else:
795 return "<%s.%s at 0x%x; %s>" % (
796 self.__module__,
797 self.__class__.__name__,
798 id(self),
799 friendly,
800 )
803class DQLDMLClauseElement(ClauseElement):
804 """represents a :class:`.ClauseElement` that compiles to a DQL or DML
805 expression, not DDL.
807 .. versionadded:: 2.0
809 """
811 if typing.TYPE_CHECKING:
813 def _compiler(self, dialect: Dialect, **kw: Any) -> SQLCompiler:
814 """Return a compiler appropriate for this ClauseElement, given a
815 Dialect."""
816 ...
818 def compile( # noqa: A001
819 self,
820 bind: Optional[_HasDialect] = None,
821 dialect: Optional[Dialect] = None,
822 **kw: Any,
823 ) -> SQLCompiler: ...
826class CompilerColumnElement(
827 roles.DMLColumnRole,
828 roles.DDLConstraintColumnRole,
829 roles.ColumnsClauseRole,
830 CompilerElement,
831):
832 """A compiler-only column element used for ad-hoc string compilations.
834 .. versionadded:: 2.0
836 """
838 __slots__ = ()
840 _propagate_attrs = util.EMPTY_DICT
841 _is_collection_aggregate = False
842 _is_implicitly_boolean = False
844 def _with_binary_element_type(self, type_):
845 raise NotImplementedError()
847 def _gen_cache_key(self, anon_map, bindparams):
848 raise NotImplementedError()
850 @property
851 def _from_objects(self) -> List[FromClause]:
852 raise NotImplementedError()
855# SQLCoreOperations should be suiting the ExpressionElementRole
856# and ColumnsClauseRole. however the MRO issues become too elaborate
857# at the moment.
858class SQLCoreOperations(Generic[_T_co], ColumnOperators, TypingOnly):
859 __slots__ = ()
861 # annotations for comparison methods
862 # these are from operators->Operators / ColumnOperators,
863 # redefined with the specific types returned by ColumnElement hierarchies
864 if typing.TYPE_CHECKING:
866 @util.non_memoized_property
867 def _propagate_attrs(self) -> _PropagateAttrsType: ...
869 def operate(
870 self, op: OperatorType, *other: Any, **kwargs: Any
871 ) -> ColumnElement[Any]: ...
873 def reverse_operate(
874 self, op: OperatorType, other: Any, **kwargs: Any
875 ) -> ColumnElement[Any]: ...
877 @overload
878 def op(
879 self,
880 opstring: str,
881 precedence: int = ...,
882 is_comparison: bool = ...,
883 *,
884 return_type: _TypeEngineArgument[_OPT],
885 python_impl: Optional[Callable[..., Any]] = None,
886 operator_class: OperatorClass = ...,
887 visit_name: Optional[str] = ...,
888 ) -> Callable[[Any], BinaryExpression[_OPT]]: ...
890 @overload
891 def op(
892 self,
893 opstring: str,
894 precedence: int = ...,
895 is_comparison: bool = ...,
896 return_type: Optional[_TypeEngineArgument[Any]] = ...,
897 python_impl: Optional[Callable[..., Any]] = ...,
898 operator_class: OperatorClass = ...,
899 visit_name: Optional[str] = ...,
900 ) -> Callable[[Any], BinaryExpression[Any]]: ...
902 def op(
903 self,
904 opstring: str,
905 precedence: int = 0,
906 is_comparison: bool = False,
907 return_type: Optional[_TypeEngineArgument[Any]] = None,
908 python_impl: Optional[Callable[..., Any]] = None,
909 operator_class: OperatorClass = OperatorClass.BASE,
910 visit_name: Optional[str] = None,
911 ) -> Callable[[Any], BinaryExpression[Any]]: ...
913 def bool_op(
914 self,
915 opstring: str,
916 precedence: int = 0,
917 python_impl: Optional[Callable[..., Any]] = None,
918 ) -> Callable[[Any], BinaryExpression[bool]]: ...
920 def __and__(self, other: Any) -> BooleanClauseList: ...
922 def __or__(self, other: Any) -> BooleanClauseList: ...
924 def __invert__(self) -> ColumnElement[_T_co]: ...
926 def __lt__(self, other: Any) -> ColumnElement[bool]: ...
928 def __le__(self, other: Any) -> ColumnElement[bool]: ...
930 # declare also that this class has an hash method otherwise
931 # it may be assumed to be None by type checkers since the
932 # object defines __eq__ and python sets it to None in that case:
933 # https://docs.python.org/3/reference/datamodel.html#object.__hash__
934 def __hash__(self) -> int: ...
936 def __eq__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
937 ...
939 def __ne__(self, other: Any) -> ColumnElement[bool]: # type: ignore[override] # noqa: E501
940 ...
942 def is_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
944 def is_not_distinct_from(self, other: Any) -> ColumnElement[bool]: ...
946 def __gt__(self, other: Any) -> ColumnElement[bool]: ...
948 def __ge__(self, other: Any) -> ColumnElement[bool]: ...
950 def __neg__(self) -> UnaryExpression[_T_co]: ...
952 def __contains__(self, other: Any) -> ColumnElement[bool]: ...
954 def __getitem__(self, index: Any) -> ColumnElement[Any]: ...
956 @overload
957 def __lshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
959 @overload
960 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
962 def __lshift__(self, other: Any) -> ColumnElement[Any]: ...
964 @overload
965 def __rlshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
967 @overload
968 def __rlshift__(self, other: Any) -> ColumnElement[Any]: ...
970 def __rlshift__(self, other: Any) -> ColumnElement[Any]: ...
972 @overload
973 def __rshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
975 @overload
976 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
978 def __rshift__(self, other: Any) -> ColumnElement[Any]: ...
980 @overload
981 def __rrshift__(self: _SQO[int], other: Any) -> ColumnElement[int]: ...
983 @overload
984 def __rrshift__(self, other: Any) -> ColumnElement[Any]: ...
986 def __rrshift__(self, other: Any) -> ColumnElement[Any]: ...
988 def __matmul__(self, other: Any) -> ColumnElement[Any]: ...
990 def __rmatmul__(self, other: Any) -> ColumnElement[Any]: ...
992 @overload
993 def concat(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
995 @overload
996 def concat(self, other: Any) -> ColumnElement[Any]: ...
998 def concat(self, other: Any) -> ColumnElement[Any]: ...
1000 def like(
1001 self, other: Any, escape: Optional[str] = None
1002 ) -> BinaryExpression[bool]: ...
1004 def ilike(
1005 self, other: Any, escape: Optional[str] = None
1006 ) -> BinaryExpression[bool]: ...
1008 def bitwise_xor(self, other: Any) -> BinaryExpression[Any]: ...
1010 def bitwise_or(self, other: Any) -> BinaryExpression[Any]: ...
1012 def bitwise_and(self, other: Any) -> BinaryExpression[Any]: ...
1014 def bitwise_not(self) -> UnaryExpression[_T_co]: ...
1016 def bitwise_lshift(self, other: Any) -> BinaryExpression[Any]: ...
1018 def bitwise_rshift(self, other: Any) -> BinaryExpression[Any]: ...
1020 def in_(
1021 self,
1022 other: Union[
1023 Iterable[Any], BindParameter[Any], roles.InElementRole
1024 ],
1025 ) -> BinaryExpression[bool]: ...
1027 def not_in(
1028 self,
1029 other: Union[
1030 Iterable[Any], BindParameter[Any], roles.InElementRole
1031 ],
1032 ) -> BinaryExpression[bool]: ...
1034 def notin_(
1035 self,
1036 other: Union[
1037 Iterable[Any], BindParameter[Any], roles.InElementRole
1038 ],
1039 ) -> BinaryExpression[bool]: ...
1041 def not_like(
1042 self, other: Any, escape: Optional[str] = None
1043 ) -> BinaryExpression[bool]: ...
1045 def notlike(
1046 self, other: Any, escape: Optional[str] = None
1047 ) -> BinaryExpression[bool]: ...
1049 def not_ilike(
1050 self, other: Any, escape: Optional[str] = None
1051 ) -> BinaryExpression[bool]: ...
1053 def notilike(
1054 self, other: Any, escape: Optional[str] = None
1055 ) -> BinaryExpression[bool]: ...
1057 def is_(self, other: Any) -> BinaryExpression[bool]: ...
1059 def is_not(self, other: Any) -> BinaryExpression[bool]: ...
1061 def isnot(self, other: Any) -> BinaryExpression[bool]: ...
1063 def startswith(
1064 self,
1065 other: Any,
1066 escape: Optional[str] = None,
1067 autoescape: bool = False,
1068 ) -> ColumnElement[bool]: ...
1070 def istartswith(
1071 self,
1072 other: Any,
1073 escape: Optional[str] = None,
1074 autoescape: bool = False,
1075 ) -> ColumnElement[bool]: ...
1077 def endswith(
1078 self,
1079 other: Any,
1080 escape: Optional[str] = None,
1081 autoescape: bool = False,
1082 ) -> ColumnElement[bool]: ...
1084 def iendswith(
1085 self,
1086 other: Any,
1087 escape: Optional[str] = None,
1088 autoescape: bool = False,
1089 ) -> ColumnElement[bool]: ...
1091 def contains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1093 def icontains(self, other: Any, **kw: Any) -> ColumnElement[bool]: ...
1095 def match(self, other: Any, **kwargs: Any) -> ColumnElement[bool]: ...
1097 def regexp_match(
1098 self, pattern: Any, flags: Optional[str] = None
1099 ) -> ColumnElement[bool]: ...
1101 def regexp_replace(
1102 self, pattern: Any, replacement: Any, flags: Optional[str] = None
1103 ) -> ColumnElement[str]: ...
1105 def desc(self) -> UnaryExpression[_T_co]: ...
1107 def asc(self) -> UnaryExpression[_T_co]: ...
1109 def nulls_first(self) -> UnaryExpression[_T_co]: ...
1111 def nullsfirst(self) -> UnaryExpression[_T_co]: ...
1113 def nulls_last(self) -> UnaryExpression[_T_co]: ...
1115 def nullslast(self) -> UnaryExpression[_T_co]: ...
1117 def collate(self, collation: str) -> CollationClause: ...
1119 def between(
1120 self, cleft: Any, cright: Any, symmetric: bool = False
1121 ) -> BinaryExpression[bool]: ...
1123 def distinct(self: _SQO[_T_co]) -> UnaryExpression[_T_co]: ...
1125 def any_(self) -> CollectionAggregate[Any]: ...
1127 def all_(self) -> CollectionAggregate[Any]: ...
1129 # numeric overloads. These need more tweaking
1130 # in particular they all need to have a variant for Optional[_T]
1131 # because Optional only applies to the data side, not the expression
1132 # side
1134 @overload
1135 def __add__(
1136 self: _SQO[_NMT],
1137 other: Any,
1138 ) -> ColumnElement[_NMT]: ...
1140 @overload
1141 def __add__(
1142 self: _SQO[str],
1143 other: Any,
1144 ) -> ColumnElement[str]: ...
1146 @overload
1147 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1149 def __add__(self, other: Any) -> ColumnElement[Any]: ...
1151 @overload
1152 def __radd__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1154 @overload
1155 def __radd__(self: _SQO[str], other: Any) -> ColumnElement[str]: ...
1157 def __radd__(self, other: Any) -> ColumnElement[Any]: ...
1159 @overload
1160 def __sub__(
1161 self: _SQO[_NMT],
1162 other: Any,
1163 ) -> ColumnElement[_NMT]: ...
1165 @overload
1166 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1168 def __sub__(self, other: Any) -> ColumnElement[Any]: ...
1170 @overload
1171 def __rsub__(
1172 self: _SQO[_NMT],
1173 other: Any,
1174 ) -> ColumnElement[_NMT]: ...
1176 @overload
1177 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1179 def __rsub__(self, other: Any) -> ColumnElement[Any]: ...
1181 @overload
1182 def __mul__(
1183 self: _SQO[_NMT],
1184 other: Any,
1185 ) -> ColumnElement[_NMT]: ...
1187 @overload
1188 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1190 def __mul__(self, other: Any) -> ColumnElement[Any]: ...
1192 @overload
1193 def __rmul__(
1194 self: _SQO[_NMT],
1195 other: Any,
1196 ) -> ColumnElement[_NMT]: ...
1198 @overload
1199 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1201 def __rmul__(self, other: Any) -> ColumnElement[Any]: ...
1203 @overload
1204 def __mod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1206 @overload
1207 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1209 def __mod__(self, other: Any) -> ColumnElement[Any]: ...
1211 @overload
1212 def __rmod__(self: _SQO[_NMT], other: Any) -> ColumnElement[_NMT]: ...
1214 @overload
1215 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1217 def __rmod__(self, other: Any) -> ColumnElement[Any]: ...
1219 @overload
1220 def __truediv__(
1221 self: _SQO[int], other: Any
1222 ) -> ColumnElement[_NUMERIC]: ...
1224 @overload
1225 def __truediv__(self: _SQO[_NT], other: Any) -> ColumnElement[_NT]: ...
1227 @overload
1228 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1230 def __truediv__(self, other: Any) -> ColumnElement[Any]: ...
1232 @overload
1233 def __rtruediv__(
1234 self: _SQO[_NMT], other: Any
1235 ) -> ColumnElement[_NUMERIC]: ...
1237 @overload
1238 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1240 def __rtruediv__(self, other: Any) -> ColumnElement[Any]: ...
1242 @overload
1243 def __floordiv__(
1244 self: _SQO[_NMT], other: Any
1245 ) -> ColumnElement[_NMT]: ...
1247 @overload
1248 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1250 def __floordiv__(self, other: Any) -> ColumnElement[Any]: ...
1252 @overload
1253 def __rfloordiv__(
1254 self: _SQO[_NMT], other: Any
1255 ) -> ColumnElement[_NMT]: ...
1257 @overload
1258 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1260 def __rfloordiv__(self, other: Any) -> ColumnElement[Any]: ...
1263class SQLColumnExpression(
1264 SQLCoreOperations[_T_co], roles.ExpressionElementRole[_T_co], TypingOnly
1265):
1266 """A type that may be used to indicate any SQL column element or object
1267 that acts in place of one.
1269 :class:`.SQLColumnExpression` is a base of
1270 :class:`.ColumnElement`, as well as within the bases of ORM elements
1271 such as :class:`.InstrumentedAttribute`, and may be used in :pep:`484`
1272 typing to indicate arguments or return values that should behave
1273 as column expressions.
1275 .. versionadded:: 2.0.0b4
1278 """
1280 __slots__ = ()
1283_SQO = SQLCoreOperations
1286class ColumnElement(
1287 roles.ColumnArgumentOrKeyRole,
1288 roles.StatementOptionRole,
1289 roles.WhereHavingRole,
1290 roles.BinaryElementRole[_T],
1291 roles.OrderByRole,
1292 roles.ColumnsClauseRole,
1293 roles.LimitOffsetRole,
1294 roles.DMLColumnRole,
1295 roles.DDLConstraintColumnRole,
1296 roles.DDLExpressionRole,
1297 SQLColumnExpression[_T],
1298 DQLDMLClauseElement,
1299):
1300 """Represent a column-oriented SQL expression suitable for usage in the
1301 "columns" clause, WHERE clause etc. of a statement.
1303 While the most familiar kind of :class:`_expression.ColumnElement` is the
1304 :class:`_schema.Column` object, :class:`_expression.ColumnElement`
1305 serves as the basis
1306 for any unit that may be present in a SQL expression, including
1307 the expressions themselves, SQL functions, bound parameters,
1308 literal expressions, keywords such as ``NULL``, etc.
1309 :class:`_expression.ColumnElement`
1310 is the ultimate base class for all such elements.
1312 A wide variety of SQLAlchemy Core functions work at the SQL expression
1313 level, and are intended to accept instances of
1314 :class:`_expression.ColumnElement` as
1315 arguments. These functions will typically document that they accept a
1316 "SQL expression" as an argument. What this means in terms of SQLAlchemy
1317 usually refers to an input which is either already in the form of a
1318 :class:`_expression.ColumnElement` object,
1319 or a value which can be **coerced** into
1320 one. The coercion rules followed by most, but not all, SQLAlchemy Core
1321 functions with regards to SQL expressions are as follows:
1323 * a literal Python value, such as a string, integer or floating
1324 point value, boolean, datetime, ``Decimal`` object, or virtually
1325 any other Python object, will be coerced into a "literal bound
1326 value". This generally means that a :func:`.bindparam` will be
1327 produced featuring the given value embedded into the construct; the
1328 resulting :class:`.BindParameter` object is an instance of
1329 :class:`_expression.ColumnElement`.
1330 The Python value will ultimately be sent
1331 to the DBAPI at execution time as a parameterized argument to the
1332 ``execute()`` or ``executemany()`` methods, after SQLAlchemy
1333 type-specific converters (e.g. those provided by any associated
1334 :class:`.TypeEngine` objects) are applied to the value.
1336 * any special object value, typically ORM-level constructs, which
1337 feature an accessor called ``__clause_element__()``. The Core
1338 expression system looks for this method when an object of otherwise
1339 unknown type is passed to a function that is looking to coerce the
1340 argument into a :class:`_expression.ColumnElement` and sometimes a
1341 :class:`_expression.SelectBase` expression.
1342 It is used within the ORM to
1343 convert from ORM-specific objects like mapped classes and
1344 mapped attributes into Core expression objects.
1346 * The Python ``None`` value is typically interpreted as ``NULL``,
1347 which in SQLAlchemy Core produces an instance of :func:`.null`.
1349 A :class:`_expression.ColumnElement` provides the ability to generate new
1350 :class:`_expression.ColumnElement`
1351 objects using Python expressions. This means that Python operators
1352 such as ``==``, ``!=`` and ``<`` are overloaded to mimic SQL operations,
1353 and allow the instantiation of further :class:`_expression.ColumnElement`
1354 instances
1355 which are composed from other, more fundamental
1356 :class:`_expression.ColumnElement`
1357 objects. For example, two :class:`.ColumnClause` objects can be added
1358 together with the addition operator ``+`` to produce
1359 a :class:`.BinaryExpression`.
1360 Both :class:`.ColumnClause` and :class:`.BinaryExpression` are subclasses
1361 of :class:`_expression.ColumnElement`:
1363 .. sourcecode:: pycon+sql
1365 >>> from sqlalchemy.sql import column
1366 >>> column("a") + column("b")
1367 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
1368 >>> print(column("a") + column("b"))
1369 {printsql}a + b
1371 .. seealso::
1373 :class:`_schema.Column`
1375 :func:`_expression.column`
1377 """
1379 __visit_name__ = "column_element"
1381 primary_key: bool = False
1382 _is_clone_of: Optional[ColumnElement[_T]]
1383 _is_column_element = True
1384 _insert_sentinel: bool = False
1385 _omit_from_statements = False
1386 _is_collection_aggregate = False
1388 foreign_keys: AbstractSet[ForeignKey] = frozenset()
1390 @util.memoized_property
1391 def _proxies(self) -> List[ColumnElement[Any]]:
1392 return []
1394 @util.non_memoized_property
1395 def _tq_label(self) -> Optional[str]:
1396 """The named label that can be used to target
1397 this column in a result set in a "table qualified" context.
1399 This label is almost always the label used when
1400 rendering <expr> AS <label> in a SELECT statement when using
1401 the LABEL_STYLE_TABLENAME_PLUS_COL label style, which is what the
1402 legacy ORM ``Query`` object uses as well.
1404 For a regular Column bound to a Table, this is typically the label
1405 <tablename>_<columnname>. For other constructs, different rules
1406 may apply, such as anonymized labels and others.
1408 .. versionchanged:: 1.4.21 renamed from ``._label``
1410 """
1411 return None
1413 key: Optional[str] = None
1414 """The 'key' that in some circumstances refers to this object in a
1415 Python namespace.
1417 This typically refers to the "key" of the column as present in the
1418 ``.c`` collection of a selectable, e.g. ``sometable.c["somekey"]`` would
1419 return a :class:`_schema.Column` with a ``.key`` of "somekey".
1421 """
1423 @HasMemoized.memoized_attribute
1424 def _tq_key_label(self) -> Optional[str]:
1425 """A label-based version of 'key' that in some circumstances refers
1426 to this object in a Python namespace.
1429 _tq_key_label comes into play when a select() statement is constructed
1430 with apply_labels(); in this case, all Column objects in the ``.c``
1431 collection are rendered as <tablename>_<columnname> in SQL; this is
1432 essentially the value of ._label. But to locate those columns in the
1433 ``.c`` collection, the name is along the lines of <tablename>_<key>;
1434 that's the typical value of .key_label.
1436 .. versionchanged:: 1.4.21 renamed from ``._key_label``
1438 """
1439 return self._proxy_key
1441 @property
1442 def _key_label(self) -> Optional[str]:
1443 """legacy; renamed to _tq_key_label"""
1444 return self._tq_key_label
1446 @property
1447 def _label(self) -> Optional[str]:
1448 """legacy; renamed to _tq_label"""
1449 return self._tq_label
1451 @property
1452 def _non_anon_label(self) -> Optional[str]:
1453 """the 'name' that naturally applies this element when rendered in
1454 SQL.
1456 Concretely, this is the "name" of a column or a label in a
1457 SELECT statement; ``<columnname>`` and ``<labelname>`` below:
1459 .. sourcecode:: sql
1461 SELECT <columnmame> FROM table
1463 SELECT column AS <labelname> FROM table
1465 Above, the two names noted will be what's present in the DBAPI
1466 ``cursor.description`` as the names.
1468 If this attribute returns ``None``, it means that the SQL element as
1469 written does not have a 100% fully predictable "name" that would appear
1470 in the ``cursor.description``. Examples include SQL functions, CAST
1471 functions, etc. While such things do return names in
1472 ``cursor.description``, they are only predictable on a
1473 database-specific basis; e.g. an expression like ``MAX(table.col)`` may
1474 appear as the string ``max`` on one database (like PostgreSQL) or may
1475 appear as the whole expression ``max(table.col)`` on SQLite.
1477 The default implementation looks for a ``.name`` attribute on the
1478 object, as has been the precedent established in SQLAlchemy for many
1479 years. An exception is made on the ``FunctionElement`` subclass
1480 so that the return value is always ``None``.
1482 .. versionadded:: 1.4.21
1486 """
1487 return getattr(self, "name", None)
1489 _render_label_in_columns_clause = True
1490 """A flag used by select._columns_plus_names that helps to determine
1491 we are actually going to render in terms of "SELECT <col> AS <label>".
1492 This flag can be returned as False for some Column objects that want
1493 to be rendered as simple "SELECT <col>"; typically columns that don't have
1494 any parent table and are named the same as what the label would be
1495 in any case.
1497 """
1499 _allow_label_resolve = True
1500 """A flag that can be flipped to prevent a column from being resolvable
1501 by string label name.
1503 The joined eager loader strategy in the ORM uses this, for example.
1505 """
1507 _is_implicitly_boolean = False
1509 _alt_names: Sequence[str] = ()
1511 if TYPE_CHECKING:
1513 def _ungroup(self) -> ColumnElement[_T]: ...
1515 @overload
1516 def self_group(self, against: None = None) -> ColumnElement[_T]: ...
1518 @overload
1519 def self_group(
1520 self, against: Optional[OperatorType] = None
1521 ) -> ColumnElement[Any]: ...
1523 def self_group(
1524 self, against: Optional[OperatorType] = None
1525 ) -> ColumnElement[Any]:
1526 if (
1527 against in (operators.and_, operators.or_, operators._asbool)
1528 and self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity
1529 ):
1530 return AsBoolean(self, operators.is_true, operators.is_false)
1531 elif against in (operators.any_op, operators.all_op):
1532 return Grouping(self)
1533 else:
1534 return self
1536 @overload
1537 def _negate(self: ColumnElement[bool]) -> ColumnElement[bool]: ...
1539 @overload
1540 def _negate(self: ColumnElement[_T]) -> ColumnElement[_T]: ...
1542 def _negate(self) -> ColumnElement[Any]:
1543 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
1544 return AsBoolean(self, operators.is_false, operators.is_true)
1545 else:
1546 grouped = self.self_group(against=operators.inv)
1547 assert isinstance(grouped, ColumnElement)
1548 return UnaryExpression(
1549 grouped,
1550 operator=operators.inv,
1551 )
1553 type: TypeEngine[_T]
1555 if not TYPE_CHECKING:
1557 @util.memoized_property
1558 def type(self) -> TypeEngine[_T]: # noqa: A001
1559 # used for delayed setup of
1560 # type_api
1561 return type_api.NULLTYPE
1563 @HasMemoized.memoized_attribute
1564 def comparator(self) -> TypeEngine.Comparator[_T]:
1565 try:
1566 comparator_factory = self.type.comparator_factory
1567 except AttributeError as err:
1568 raise TypeError(
1569 "Object %r associated with '.type' attribute "
1570 "is not a TypeEngine class or object" % self.type
1571 ) from err
1572 else:
1573 return comparator_factory(self)
1575 def __setstate__(self, state):
1576 self.__dict__.update(state)
1578 def __getattr__(self, key: str) -> Any:
1579 try:
1580 return getattr(self.comparator, key)
1581 except AttributeError as err:
1582 raise AttributeError(
1583 "Neither %r object nor %r object has an attribute %r"
1584 % (
1585 type(self).__name__,
1586 type(self.comparator).__name__,
1587 key,
1588 )
1589 ) from err
1591 def operate(
1592 self,
1593 op: operators.OperatorType,
1594 *other: Any,
1595 **kwargs: Any,
1596 ) -> ColumnElement[Any]:
1597 return op(self.comparator, *other, **kwargs) # type: ignore[no-any-return] # noqa: E501
1599 def reverse_operate(
1600 self, op: operators.OperatorType, other: Any, **kwargs: Any
1601 ) -> ColumnElement[Any]:
1602 return op(other, self.comparator, **kwargs) # type: ignore[no-any-return] # noqa: E501
1604 def _bind_param(
1605 self,
1606 operator: operators.OperatorType,
1607 obj: Any,
1608 type_: Optional[TypeEngine[_T]] = None,
1609 expanding: bool = False,
1610 ) -> BindParameter[_T]:
1611 return BindParameter(
1612 None,
1613 obj,
1614 _compared_to_operator=operator,
1615 type_=type_,
1616 _compared_to_type=self.type,
1617 unique=True,
1618 expanding=expanding,
1619 )
1621 @property
1622 def expression(self) -> ColumnElement[Any]:
1623 """Return a column expression.
1625 Part of the inspection interface; returns self.
1627 """
1628 return self
1630 @property
1631 def _select_iterable(self) -> _SelectIterable:
1632 return (self,)
1634 @util.memoized_property
1635 def base_columns(self) -> FrozenSet[ColumnElement[Any]]:
1636 return frozenset(c for c in self.proxy_set if not c._proxies)
1638 @util.memoized_property
1639 def proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1640 """set of all columns we are proxying
1642 as of 2.0 this is explicitly deannotated columns. previously it was
1643 effectively deannotated columns but wasn't enforced. annotated
1644 columns should basically not go into sets if at all possible because
1645 their hashing behavior is very non-performant.
1647 """
1648 return frozenset([self._deannotate()]).union(
1649 itertools.chain(*[c.proxy_set for c in self._proxies])
1650 )
1652 @util.memoized_property
1653 def _expanded_proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1654 return frozenset(_expand_cloned(self.proxy_set))
1656 def _uncached_proxy_list(self) -> List[ColumnElement[Any]]:
1657 """An 'uncached' version of proxy set.
1659 This list includes annotated columns which perform very poorly in
1660 set operations.
1662 """
1664 return [self] + list(
1665 itertools.chain(*[c._uncached_proxy_list() for c in self._proxies])
1666 )
1668 def shares_lineage(self, othercolumn: ColumnElement[Any]) -> bool:
1669 """Return True if the given :class:`_expression.ColumnElement`
1670 has a common ancestor to this :class:`_expression.ColumnElement`."""
1672 return bool(self.proxy_set.intersection(othercolumn.proxy_set))
1674 def _compare_name_for_result(self, other: ColumnElement[Any]) -> bool:
1675 """Return True if the given column element compares to this one
1676 when targeting within a result row."""
1678 return (
1679 hasattr(other, "name")
1680 and hasattr(self, "name")
1681 and other.name == self.name
1682 )
1684 @HasMemoized.memoized_attribute
1685 def _proxy_key(self) -> Optional[str]:
1686 if self._annotations and "proxy_key" in self._annotations:
1687 return cast(str, self._annotations["proxy_key"])
1689 name = self.key
1690 if not name:
1691 # there's a bit of a seeming contradiction which is that the
1692 # "_non_anon_label" of a column can in fact be an
1693 # "_anonymous_label"; this is when it's on a column that is
1694 # proxying for an anonymous expression in a subquery.
1695 name = self._non_anon_label
1697 if isinstance(name, _anonymous_label):
1698 return None
1699 else:
1700 return name
1702 @HasMemoized.memoized_attribute
1703 def _expression_label(self) -> Optional[str]:
1704 """a suggested label to use in the case that the column has no name,
1705 which should be used if possible as the explicit 'AS <label>'
1706 where this expression would normally have an anon label.
1708 this is essentially mostly what _proxy_key does except it returns
1709 None if the column has a normal name that can be used.
1711 """
1713 if getattr(self, "name", None) is not None:
1714 return None
1715 elif self._annotations and "proxy_key" in self._annotations:
1716 return cast(str, self._annotations["proxy_key"])
1717 else:
1718 return None
1720 def _make_proxy(
1721 self,
1722 selectable: FromClause,
1723 *,
1724 primary_key: ColumnSet,
1725 foreign_keys: Set[KeyedColumnElement[Any]],
1726 name: Optional[str] = None,
1727 key: Optional[str] = None,
1728 name_is_truncatable: bool = False,
1729 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
1730 **kw: Any,
1731 ) -> typing_Tuple[str, ColumnClause[_T]]:
1732 """Create a new :class:`_expression.ColumnElement` representing this
1733 :class:`_expression.ColumnElement` as it appears in the select list of
1734 a descending selectable.
1736 """
1737 if name is None:
1738 name = self._anon_name_label
1739 if key is None:
1740 key = self._proxy_key
1741 else:
1742 key = name
1744 assert key is not None
1746 co: ColumnClause[_T] = ColumnClause(
1747 (
1748 coercions.expect(roles.TruncatedLabelRole, name)
1749 if name_is_truncatable
1750 else name
1751 ),
1752 type_=getattr(self, "type", None),
1753 _selectable=selectable,
1754 )
1756 co._propagate_attrs = selectable._propagate_attrs
1757 if compound_select_cols:
1758 co._proxies = list(compound_select_cols)
1759 else:
1760 co._proxies = [self]
1761 if selectable._is_clone_of is not None:
1762 co._is_clone_of = selectable._is_clone_of.columns.get(key)
1763 return key, co
1765 def cast(self, type_: _TypeEngineArgument[_OPT]) -> Cast[_OPT]:
1766 """Produce a type cast, i.e. ``CAST(<expression> AS <type>)``.
1768 This is a shortcut to the :func:`_expression.cast` function.
1770 .. seealso::
1772 :ref:`tutorial_casts`
1774 :func:`_expression.cast`
1776 :func:`_expression.type_coerce`
1778 """
1779 return Cast(self, type_)
1781 def label(self, name: Optional[str]) -> Label[_T]:
1782 """Produce a column label, i.e. ``<columnname> AS <name>``.
1784 This is a shortcut to the :func:`_expression.label` function.
1786 If 'name' is ``None``, an anonymous label name will be generated.
1788 """
1789 return Label(name, self, self.type)
1791 def _anon_label(
1792 self, seed: Optional[str], add_hash: Optional[int] = None
1793 ) -> _anonymous_label:
1794 while self._is_clone_of is not None:
1795 self = self._is_clone_of
1797 # as of 1.4 anonymous label for ColumnElement uses hash(), not id(),
1798 # as the identifier, because a column and its annotated version are
1799 # the same thing in a SQL statement
1800 hash_value = hash(self)
1802 if add_hash:
1803 # this path is used for disambiguating anon labels that would
1804 # otherwise be the same name for the same element repeated.
1805 # an additional numeric value is factored in for each label.
1807 # shift hash(self) (which is id(self), typically 8 byte integer)
1808 # 16 bits leftward. fill extra add_hash on right
1809 assert add_hash < (2 << 15)
1810 assert seed
1811 hash_value = (hash_value << 16) | add_hash
1813 # extra underscore is added for labels with extra hash
1814 # values, to isolate the "deduped anon" namespace from the
1815 # regular namespace. eliminates chance of these
1816 # manufactured hash values overlapping with regular ones for some
1817 # undefined python interpreter
1818 seed = seed + "_"
1820 if isinstance(seed, _anonymous_label):
1821 # NOTE: the space after the hash is required
1822 return _anonymous_label(f"{seed}%({hash_value} )s")
1824 return _anonymous_label.safe_construct(hash_value, seed or "anon")
1826 @util.memoized_property
1827 def _anon_name_label(self) -> str:
1828 """Provides a constant 'anonymous label' for this ColumnElement.
1830 This is a label() expression which will be named at compile time.
1831 The same label() is returned each time ``anon_label`` is called so
1832 that expressions can reference ``anon_label`` multiple times,
1833 producing the same label name at compile time.
1835 The compiler uses this function automatically at compile time
1836 for expressions that are known to be 'unnamed' like binary
1837 expressions and function calls.
1839 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1840 public and is renamed to _anon_name_label. anon_name exists
1841 for backwards compat
1843 """
1844 name = getattr(self, "name", None)
1845 return self._anon_label(name)
1847 @util.memoized_property
1848 def _anon_key_label(self) -> _anonymous_label:
1849 """Provides a constant 'anonymous key label' for this ColumnElement.
1851 Compare to ``anon_label``, except that the "key" of the column,
1852 if available, is used to generate the label.
1854 This is used when a deduplicating key is placed into the columns
1855 collection of a selectable.
1857 .. versionchanged:: 1.4.9 - this attribute was not intended to be
1858 public and is renamed to _anon_key_label. anon_key_label exists
1859 for backwards compat
1861 """
1862 return self._anon_label(self._proxy_key)
1864 @property
1865 @util.deprecated(
1866 "1.4",
1867 "The :attr:`_expression.ColumnElement.anon_label` attribute is now "
1868 "private, and the public accessor is deprecated.",
1869 )
1870 def anon_label(self) -> str:
1871 return self._anon_name_label
1873 @property
1874 @util.deprecated(
1875 "1.4",
1876 "The :attr:`_expression.ColumnElement.anon_key_label` attribute is "
1877 "now private, and the public accessor is deprecated.",
1878 )
1879 def anon_key_label(self) -> str:
1880 return self._anon_key_label
1882 def _dedupe_anon_label_idx(self, idx: int) -> str:
1883 """label to apply to a column that is anon labeled, but repeated
1884 in the SELECT, so that we have to make an "extra anon" label that
1885 disambiguates it from the previous appearance.
1887 these labels come out like "foo_bar_id__1" and have double underscores
1888 in them.
1890 """
1891 label = getattr(self, "name", None)
1893 # current convention is that if the element doesn't have a
1894 # ".name" (usually because it is not NamedColumn), we try to
1895 # use a "table qualified" form for the "dedupe anon" label,
1896 # based on the notion that a label like
1897 # "CAST(casttest.v1 AS DECIMAL) AS casttest_v1__1" looks better than
1898 # "CAST(casttest.v1 AS DECIMAL) AS anon__1"
1900 if label is None:
1901 return self._dedupe_anon_tq_label_idx(idx)
1902 else:
1903 return self._anon_label(label, add_hash=idx)
1905 @util.memoized_property
1906 def _anon_tq_label(self) -> _anonymous_label:
1907 return self._anon_label(getattr(self, "_tq_label", None))
1909 @util.memoized_property
1910 def _anon_tq_key_label(self) -> _anonymous_label:
1911 return self._anon_label(getattr(self, "_tq_key_label", None))
1913 def _dedupe_anon_tq_label_idx(self, idx: int) -> _anonymous_label:
1914 label = getattr(self, "_tq_label", None) or "anon"
1916 return self._anon_label(label, add_hash=idx)
1919class KeyedColumnElement(ColumnElement[_T]):
1920 """ColumnElement where ``.key`` is non-None."""
1922 _is_keyed_column_element = True
1924 key: str
1927class WrapsColumnExpression(ColumnElement[_T]):
1928 """Mixin that defines a :class:`_expression.ColumnElement`
1929 as a wrapper with special
1930 labeling behavior for an expression that already has a name.
1932 .. versionadded:: 1.4
1934 .. seealso::
1936 :ref:`change_4449`
1939 """
1941 @property
1942 def wrapped_column_expression(self) -> ColumnElement[_T]:
1943 raise NotImplementedError()
1945 @util.non_memoized_property
1946 def _tq_label(self) -> Optional[str]:
1947 wce = self.wrapped_column_expression
1948 if hasattr(wce, "_tq_label"):
1949 return wce._tq_label
1950 else:
1951 return None
1953 @property
1954 def _label(self) -> Optional[str]:
1955 return self._tq_label
1957 @property
1958 def _non_anon_label(self) -> Optional[str]:
1959 return None
1961 @util.non_memoized_property
1962 def _anon_name_label(self) -> str:
1963 wce = self.wrapped_column_expression
1965 # this logic tries to get the WrappedColumnExpression to render
1966 # with "<expr> AS <name>", where "<name>" is the natural name
1967 # within the expression itself. e.g. "CAST(table.foo) AS foo".
1968 if not wce._is_text_clause:
1969 nal = wce._non_anon_label
1970 if nal:
1971 return nal
1972 elif hasattr(wce, "_anon_name_label"):
1973 return wce._anon_name_label
1974 return super()._anon_name_label
1976 def _dedupe_anon_label_idx(self, idx: int) -> str:
1977 wce = self.wrapped_column_expression
1978 nal = wce._non_anon_label
1979 if nal:
1980 return self._anon_label(nal + "_")
1981 else:
1982 return self._dedupe_anon_tq_label_idx(idx)
1984 @property
1985 def _proxy_key(self):
1986 wce = self.wrapped_column_expression
1988 if not wce._is_text_clause:
1989 return wce._proxy_key
1990 return super()._proxy_key
1993class DMLTargetCopy(roles.InElementRole, KeyedColumnElement[_T]):
1994 """Refer to another column's VALUES or SET expression in an INSERT or
1995 UPDATE statement.
1997 See the public-facing :func:`_sql.from_dml_column` constructor for
1998 background.
2000 .. versionadded:: 2.1
2003 """
2005 def __init__(self, column: _OnlyColumnArgument[_T]):
2006 self.column = coercions.expect(roles.ColumnArgumentRole, column)
2007 self.type = self.column.type
2009 __visit_name__ = "dmltargetcopy"
2011 _traverse_internals: _TraverseInternalsType = [
2012 ("column", InternalTraversal.dp_clauseelement),
2013 ]
2016class BindParameter(roles.InElementRole, KeyedColumnElement[_T]):
2017 r"""Represent a "bound expression".
2019 :class:`.BindParameter` is invoked explicitly using the
2020 :func:`.bindparam` function, as in::
2022 from sqlalchemy import bindparam
2024 stmt = select(users_table).where(
2025 users_table.c.name == bindparam("username")
2026 )
2028 Detailed discussion of how :class:`.BindParameter` is used is
2029 at :func:`.bindparam`.
2031 .. seealso::
2033 :func:`.bindparam`
2035 """
2037 __visit_name__ = "bindparam"
2039 _traverse_internals: _TraverseInternalsType = [
2040 ("key", InternalTraversal.dp_anon_name),
2041 ("type", InternalTraversal.dp_type),
2042 ("callable", InternalTraversal.dp_plain_dict),
2043 ("value", InternalTraversal.dp_plain_obj),
2044 ("literal_execute", InternalTraversal.dp_boolean),
2045 ]
2047 key: str
2048 _anon_map_key: Optional[str] = None
2049 type: TypeEngine[_T]
2050 value: Optional[_T]
2052 _is_crud = False
2053 _is_bind_parameter = True
2055 # bindparam implements its own _gen_cache_key() method however
2056 # we check subclasses for this flag, else no cache key is generated
2057 inherit_cache = True
2059 def __init__(
2060 self,
2061 key: Optional[str],
2062 value: Any = _NoArg.NO_ARG,
2063 type_: Optional[_TypeEngineArgument[_T]] = None,
2064 unique: bool = False,
2065 required: Union[bool, Literal[_NoArg.NO_ARG]] = _NoArg.NO_ARG,
2066 quote: Optional[bool] = None,
2067 callable_: Optional[Callable[[], Any]] = None,
2068 expanding: bool = False,
2069 isoutparam: bool = False,
2070 literal_execute: bool = False,
2071 _compared_to_operator: Optional[OperatorType] = None,
2072 _compared_to_type: Optional[TypeEngine[Any]] = None,
2073 _is_crud: bool = False,
2074 ):
2075 if required is _NoArg.NO_ARG:
2076 required = value is _NoArg.NO_ARG and callable_ is None
2077 if value is _NoArg.NO_ARG:
2078 value = None
2080 if quote is not None:
2081 key = quoted_name.construct(key, quote)
2083 if unique:
2084 self.key, self._anon_map_key = (
2085 _anonymous_label.safe_construct_with_key(
2086 id(self),
2087 (
2088 key
2089 if key is not None
2090 and not isinstance(key, _anonymous_label)
2091 else "param"
2092 ),
2093 sanitize_key=True,
2094 )
2095 )
2096 elif key:
2097 self.key = key
2098 else:
2099 self.key, self._anon_map_key = (
2100 _anonymous_label.safe_construct_with_key(id(self), "param")
2101 )
2103 # identifying key that won't change across
2104 # clones, used to identify the bind's logical
2105 # identity
2106 self._identifying_key = self.key
2108 # key that was passed in the first place, used to
2109 # generate new keys
2110 self._orig_key = key or "param"
2112 self.unique = unique
2113 self.value = value
2114 self.callable = callable_
2115 self.isoutparam = isoutparam
2116 self.required = required
2118 # indicate an "expanding" parameter; the compiler sets this
2119 # automatically in the compiler _render_in_expr_w_bindparam method
2120 # for an IN expression
2121 self.expanding = expanding
2123 # this is another hint to help w/ expanding and is typically
2124 # set in the compiler _render_in_expr_w_bindparam method for an
2125 # IN expression
2126 self.expand_op = None
2128 self.literal_execute = literal_execute
2129 if _is_crud:
2130 self._is_crud = True
2132 if type_ is None:
2133 if expanding:
2134 if value:
2135 check_value = value[0]
2136 else:
2137 check_value = type_api._NO_VALUE_IN_LIST
2138 else:
2139 check_value = value
2140 if _compared_to_type is not None:
2141 self.type = _compared_to_type.coerce_compared_value(
2142 _compared_to_operator, check_value
2143 )
2144 else:
2145 self.type = type_api._resolve_value_to_type(check_value)
2146 elif isinstance(type_, type):
2147 self.type = type_()
2148 elif is_tuple_type(type_):
2149 if value:
2150 if expanding:
2151 check_value = value[0]
2152 else:
2153 check_value = value
2154 cast("BindParameter[TupleAny]", self).type = (
2155 type_._resolve_values_to_types(check_value)
2156 )
2157 else:
2158 cast("BindParameter[TupleAny]", self).type = type_
2159 else:
2160 self.type = type_
2162 def _with_value(self, value, maintain_key=False, required=NO_ARG):
2163 """Return a copy of this :class:`.BindParameter` with the given value
2164 set.
2165 """
2166 cloned = self._clone(maintain_key=maintain_key)
2167 cloned.value = value
2168 cloned.callable = None
2169 cloned.required = required if required is not NO_ARG else self.required
2170 if cloned.type is type_api.NULLTYPE:
2171 cloned.type = type_api._resolve_value_to_type(value)
2172 return cloned
2174 @property
2175 def effective_value(self) -> Optional[_T]:
2176 """Return the value of this bound parameter,
2177 taking into account if the ``callable`` parameter
2178 was set.
2180 The ``callable`` value will be evaluated
2181 and returned if present, else ``value``.
2183 """
2184 if self.callable:
2185 # TODO: set up protocol for bind parameter callable
2186 return self.callable() # type: ignore
2187 else:
2188 return self.value
2190 def render_literal_execute(self) -> Self:
2191 """Produce a copy of this bound parameter that will enable the
2192 :paramref:`_sql.BindParameter.literal_execute` flag.
2194 The :paramref:`_sql.BindParameter.literal_execute` flag will
2195 have the effect of the parameter rendered in the compiled SQL
2196 string using ``[POSTCOMPILE]`` form, which is a special form that
2197 is converted to be a rendering of the literal value of the parameter
2198 at SQL execution time. The rationale is to support caching
2199 of SQL statement strings that can embed per-statement literal values,
2200 such as LIMIT and OFFSET parameters, in the final SQL string that
2201 is passed to the DBAPI. Dialects in particular may want to use
2202 this method within custom compilation schemes.
2204 .. versionadded:: 1.4.5
2206 .. seealso::
2208 :ref:`engine_thirdparty_caching`
2210 """
2211 c: Self = ClauseElement._clone(self)
2212 c.literal_execute = True
2213 return c
2215 def _negate_in_binary(self, negated_op, original_op):
2216 if self.expand_op is original_op:
2217 bind = self._clone()
2218 bind.expand_op = negated_op
2219 return bind
2220 else:
2221 return self
2223 def _with_binary_element_type(self, type_: TypeEngine[Any]) -> Self:
2224 c: Self = ClauseElement._clone(self)
2225 c.type = type_
2226 return c
2228 def _clone(self, maintain_key: bool = False, **kw: Any) -> Self:
2229 c: Self = ClauseElement._clone(self, **kw)
2230 # ensure all the BindParameter objects stay in cloned set.
2231 # in #7823, we changed "clone" so that a clone only keeps a reference
2232 # to the "original" element, since for column correspondence, that's
2233 # all we need. However, for BindParam, _cloned_set is used by
2234 # the "cache key bind match" lookup, which means if any of those
2235 # interim BindParameter objects became part of a cache key in the
2236 # cache, we need it. So here, make sure all clones keep carrying
2237 # forward.
2238 c._cloned_set.update(self._cloned_set)
2239 if not maintain_key and self.unique:
2240 c.key, c._anon_map_key = _anonymous_label.safe_construct_with_key(
2241 id(c), c._orig_key or "param", sanitize_key=True
2242 )
2243 return c
2245 def _gen_cache_key(self, anon_map, bindparams):
2246 _gen_cache_ok = self.__class__.__dict__.get("inherit_cache", False)
2248 if not _gen_cache_ok:
2249 if anon_map is not None:
2250 anon_map[NO_CACHE] = True
2251 return None
2253 id_, found = anon_map.get_anon(self)
2254 if found:
2255 return (id_, self.__class__)
2257 if bindparams is not None:
2258 bindparams.append(self)
2260 return (
2261 id_,
2262 self.__class__,
2263 self.type._static_cache_key,
2264 (
2265 anon_map[self._anon_map_key]
2266 if self._anon_map_key is not None
2267 else self.key
2268 ),
2269 self.literal_execute,
2270 )
2272 def _convert_to_unique(self):
2273 if not self.unique:
2274 self.unique = True
2275 self.key, self._anon_map_key = (
2276 _anonymous_label.safe_construct_with_key(
2277 id(self), self._orig_key or "param", sanitize_key=True
2278 )
2279 )
2281 def __getstate__(self):
2282 """execute a deferred value for serialization purposes."""
2284 d = self.__dict__.copy()
2285 v = self.value
2286 if self.callable:
2287 v = self.callable()
2288 d["callable"] = None
2289 d["value"] = v
2290 return d
2292 def __setstate__(self, state):
2293 if state.get("unique", False):
2294 anon_and_key = _anonymous_label.safe_construct_with_key(
2295 id(self), state.get("_orig_key", "param"), sanitize_key=True
2296 )
2297 state["key"], state["_anon_map_key"] = anon_and_key
2298 self.__dict__.update(state)
2300 def __repr__(self):
2301 return "%s(%r, %r, type_=%r)" % (
2302 self.__class__.__name__,
2303 self.key,
2304 self.value,
2305 self.type,
2306 )
2309class TypeClause(DQLDMLClauseElement):
2310 """Handle a type keyword in a SQL statement.
2312 Used by the ``Case`` statement.
2314 """
2316 __visit_name__ = "typeclause"
2318 _traverse_internals: _TraverseInternalsType = [
2319 ("type", InternalTraversal.dp_type)
2320 ]
2321 type: TypeEngine[Any]
2323 def __init__(self, type_: TypeEngine[Any]):
2324 self.type = type_
2327class AbstractTextClause(
2328 roles.DDLConstraintColumnRole,
2329 roles.DDLExpressionRole,
2330 roles.StatementOptionRole,
2331 roles.WhereHavingRole,
2332 roles.OrderByRole,
2333 roles.FromClauseRole,
2334 roles.SelectStatementRole,
2335 roles.InElementRole,
2336 Generative,
2337 ExecutableStatement,
2338 DQLDMLClauseElement,
2339 roles.BinaryElementRole[Any],
2340):
2341 """Base class for textual SQL constructs like TextClause and TString."""
2343 __visit_name__: str
2345 _is_text_clause = True
2346 _is_textual = True
2347 _is_implicitly_boolean = False
2348 _render_label_in_columns_clause = False
2349 _omit_from_statements = False
2350 _is_collection_aggregate = False
2352 @property
2353 def _hide_froms(self) -> Iterable[FromClause]:
2354 return ()
2356 def __and__(self, other):
2357 # support use in select.where(), query.filter()
2358 return and_(self, other)
2360 @property
2361 def _select_iterable(self) -> _SelectIterable:
2362 return (self,)
2364 # help in those cases where text/tstring() is
2365 # interpreted in a column expression situation
2366 key: Optional[str] = None
2367 _label: Optional[str] = None
2369 _allow_label_resolve = False
2371 @property
2372 def type(self) -> TypeEngine[Any]:
2373 return type_api.NULLTYPE
2375 @property
2376 def comparator(self):
2377 return self.type.comparator_factory(self) # type: ignore
2379 def self_group(
2380 self, against: Optional[OperatorType] = None
2381 ) -> Union[Self, Grouping[Any]]:
2382 if against is operators.in_op:
2383 return Grouping(self)
2384 else:
2385 return self
2387 def bindparams(
2388 self,
2389 *binds: BindParameter[Any],
2390 **names_to_values: Any,
2391 ) -> Self:
2392 """Establish the values and/or types of bound parameters within
2393 this :class:`_expression.AbstractTextClause` construct.
2395 This is implemented only for :class:`.TextClause` will raise
2396 ``NotImplementedError`` for :class:`.TString`.
2398 """
2399 raise NotImplementedError()
2401 @util.preload_module("sqlalchemy.sql.selectable")
2402 def columns(
2403 self,
2404 *cols: _OnlyColumnArgument[Any],
2405 **types: _TypeEngineArgument[Any],
2406 ) -> TextualSelect:
2407 r"""Turn this :class:`_expression.AbstractTextClause` object into a
2408 :class:`_expression.TextualSelect`
2409 object that serves the same role as a SELECT
2410 statement.
2412 The :class:`_expression.TextualSelect` is part of the
2413 :class:`_expression.SelectBase`
2414 hierarchy and can be embedded into another statement by using the
2415 :meth:`_expression.TextualSelect.subquery` method to produce a
2416 :class:`.Subquery`
2417 object, which can then be SELECTed from.
2419 This function essentially bridges the gap between an entirely
2420 textual SELECT statement and the SQL expression language concept
2421 of a "selectable"::
2423 from sqlalchemy.sql import column, text
2425 stmt = text("SELECT id, name FROM some_table")
2426 stmt = stmt.columns(column("id"), column("name")).subquery("st")
2428 stmt = (
2429 select(mytable)
2430 .select_from(mytable.join(stmt, mytable.c.name == stmt.c.name))
2431 .where(stmt.c.id > 5)
2432 )
2434 Above, we pass a series of :func:`_expression.column` elements to the
2435 :meth:`_expression.AbstractTextClause.columns` method positionally.
2436 These :func:`_expression.column` elements now become first class
2437 elements upon the :attr:`_expression.TextualSelect.selected_columns`
2438 column collection, which then become part of the :attr:`.Subquery.c`
2439 collection after :meth:`_expression.TextualSelect.subquery` is invoked.
2441 The column expressions we pass to
2442 :meth:`_expression.AbstractTextClause.columns` may also be typed; when
2443 we do so, these :class:`.TypeEngine` objects become the effective
2444 return type of the column, so that SQLAlchemy's result-set-processing
2445 systems may be used on the return values. This is often needed for
2446 types such as date or boolean types, as well as for unicode processing
2447 on some dialect configurations::
2449 stmt = text("SELECT id, name, timestamp FROM some_table")
2450 stmt = stmt.columns(
2451 column("id", Integer),
2452 column("name", Unicode),
2453 column("timestamp", DateTime),
2454 )
2456 for id, name, timestamp in connection.execute(stmt):
2457 print(id, name, timestamp)
2459 As a shortcut to the above syntax, keyword arguments referring to
2460 types alone may be used, if only type conversion is needed::
2462 stmt = text("SELECT id, name, timestamp FROM some_table")
2463 stmt = stmt.columns(id=Integer, name=Unicode, timestamp=DateTime)
2465 for id, name, timestamp in connection.execute(stmt):
2466 print(id, name, timestamp)
2468 The positional form of :meth:`_expression.AbstractTextClause.columns`
2469 also provides the unique feature of **positional column targeting**,
2470 which is particularly useful when using the ORM with complex textual
2471 queries. If we specify the columns from our model to
2472 :meth:`_expression.AbstractTextClause.columns`, the result set will
2473 match to those columns positionally, meaning the name or origin of the
2474 column in the textual SQL doesn't matter::
2476 stmt = text(
2477 "SELECT users.id, addresses.id, users.id, "
2478 "users.name, addresses.email_address AS email "
2479 "FROM users JOIN addresses ON users.id=addresses.user_id "
2480 "WHERE users.id = 1"
2481 ).columns(
2482 User.id,
2483 Address.id,
2484 Address.user_id,
2485 User.name,
2486 Address.email_address,
2487 )
2489 query = (
2490 session.query(User)
2491 .from_statement(stmt)
2492 .options(contains_eager(User.addresses))
2493 )
2495 The :meth:`_expression.AbstractTextClause.columns` method provides a
2496 direct route to calling :meth:`_expression.FromClause.subquery` as well
2497 as :meth:`_expression.SelectBase.cte` against a textual SELECT
2498 statement::
2500 stmt = stmt.columns(id=Integer, name=String).cte("st")
2502 stmt = select(sometable).where(sometable.c.id == stmt.c.id)
2504 :param \*cols: A series of :class:`_expression.ColumnElement` objects,
2505 typically
2506 :class:`_schema.Column` objects from a :class:`_schema.Table`
2507 or ORM level
2508 column-mapped attributes, representing a set of columns that this
2509 textual string will SELECT from.
2511 :param \**types: A mapping of string names to :class:`.TypeEngine`
2512 type objects indicating the datatypes to use for names that are
2513 SELECTed from the textual string. Prefer to use the ``*cols``
2514 argument as it also indicates positional ordering.
2516 """
2517 selectable = util.preloaded.sql_selectable
2519 input_cols: List[NamedColumn[Any]] = [
2520 coercions.expect(roles.LabeledColumnExprRole, col) for col in cols
2521 ]
2523 positional_input_cols = [
2524 (
2525 ColumnClause(col.key, types.pop(col.key))
2526 if col.key in types
2527 else col
2528 )
2529 for col in input_cols
2530 ]
2531 keyed_input_cols: List[NamedColumn[Any]] = [
2532 ColumnClause(key, type_) for key, type_ in types.items()
2533 ]
2535 elem = selectable.TextualSelect.__new__(selectable.TextualSelect)
2536 elem._init(
2537 self,
2538 positional_input_cols + keyed_input_cols,
2539 positional=bool(positional_input_cols) and not keyed_input_cols,
2540 )
2541 return elem
2544class TextClause(AbstractTextClause, inspection.Inspectable["TextClause"]):
2545 """Represent a literal SQL text fragment.
2547 E.g.::
2549 from sqlalchemy import text
2551 t = text("SELECT * FROM users")
2552 result = connection.execute(t)
2554 The :class:`_expression.TextClause` construct is produced using the
2555 :func:`_expression.text`
2556 function; see that function for full documentation.
2558 .. seealso::
2560 :func:`_expression.text`
2562 """
2564 __visit_name__ = "textclause"
2566 _traverse_internals: _TraverseInternalsType = [
2567 ("_bindparams", InternalTraversal.dp_string_clauseelement_dict),
2568 ("text", InternalTraversal.dp_string),
2569 ] + ExecutableStatement._executable_traverse_internals
2571 _bind_params_regex = re.compile(r"(?<![:\w\x5c]):(\w+)(?!:)", re.UNICODE)
2573 @property
2574 def _is_star(self) -> bool: # type: ignore[override]
2575 return self.text == "*"
2577 def __init__(self, text: str):
2578 self._bindparams: Dict[str, BindParameter[Any]] = {}
2580 def repl(m):
2581 self._bindparams[m.group(1)] = BindParameter(m.group(1))
2582 return ":%s" % m.group(1)
2584 # scan the string and search for bind parameter names, add them
2585 # to the list of bindparams
2586 self.text = self._bind_params_regex.sub(repl, text)
2588 @_generative
2589 def bindparams(
2590 self,
2591 *binds: BindParameter[Any],
2592 **names_to_values: Any,
2593 ) -> Self:
2594 """Establish the values and/or types of bound parameters within
2595 this :class:`_expression.TextClause` construct.
2597 Given a text construct such as::
2599 from sqlalchemy import text
2601 stmt = text(
2602 "SELECT id, name FROM user WHERE name=:name AND timestamp=:timestamp"
2603 )
2605 the :meth:`_expression.TextClause.bindparams`
2606 method can be used to establish
2607 the initial value of ``:name`` and ``:timestamp``,
2608 using simple keyword arguments::
2610 stmt = stmt.bindparams(
2611 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2612 )
2614 Where above, new :class:`.BindParameter` objects
2615 will be generated with the names ``name`` and ``timestamp``, and
2616 values of ``jack`` and ``datetime.datetime(2012, 10, 8, 15, 12, 5)``,
2617 respectively. The types will be
2618 inferred from the values given, in this case :class:`.String` and
2619 :class:`.DateTime`.
2621 When specific typing behavior is needed, the positional ``*binds``
2622 argument can be used in which to specify :func:`.bindparam` constructs
2623 directly. These constructs must include at least the ``key``
2624 argument, then an optional value and type::
2626 from sqlalchemy import bindparam
2628 stmt = stmt.bindparams(
2629 bindparam("name", value="jack", type_=String),
2630 bindparam("timestamp", type_=DateTime),
2631 )
2633 Above, we specified the type of :class:`.DateTime` for the
2634 ``timestamp`` bind, and the type of :class:`.String` for the ``name``
2635 bind. In the case of ``name`` we also set the default value of
2636 ``"jack"``.
2638 Additional bound parameters can be supplied at statement execution
2639 time, e.g.::
2641 result = connection.execute(
2642 stmt, timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2643 )
2645 The :meth:`_expression.TextClause.bindparams`
2646 method can be called repeatedly,
2647 where it will reuse existing :class:`.BindParameter` objects to add
2648 new information. For example, we can call
2649 :meth:`_expression.TextClause.bindparams`
2650 first with typing information, and a
2651 second time with value information, and it will be combined::
2653 stmt = text(
2654 "SELECT id, name FROM user WHERE name=:name "
2655 "AND timestamp=:timestamp"
2656 )
2657 stmt = stmt.bindparams(
2658 bindparam("name", type_=String), bindparam("timestamp", type_=DateTime)
2659 )
2660 stmt = stmt.bindparams(
2661 name="jack", timestamp=datetime.datetime(2012, 10, 8, 15, 12, 5)
2662 )
2664 The :meth:`_expression.TextClause.bindparams`
2665 method also supports the concept of
2666 **unique** bound parameters. These are parameters that are
2667 "uniquified" on name at statement compilation time, so that multiple
2668 :func:`_expression.text`
2669 constructs may be combined together without the names
2670 conflicting. To use this feature, specify the
2671 :paramref:`.BindParameter.unique` flag on each :func:`.bindparam`
2672 object::
2674 stmt1 = text("select id from table where name=:name").bindparams(
2675 bindparam("name", value="name1", unique=True)
2676 )
2677 stmt2 = text("select id from table where name=:name").bindparams(
2678 bindparam("name", value="name2", unique=True)
2679 )
2681 union = union_all(stmt1.columns(column("id")), stmt2.columns(column("id")))
2683 The above statement will render as:
2685 .. sourcecode:: sql
2687 select id from table where name=:name_1
2688 UNION ALL select id from table where name=:name_2
2690 """ # noqa: E501
2691 self._bindparams = new_params = self._bindparams.copy()
2693 for bind in binds:
2694 try:
2695 # the regex used for text() currently will not match
2696 # a unique/anonymous key in any case, so use the _orig_key
2697 # so that a text() construct can support unique parameters
2698 existing = new_params[bind._orig_key]
2699 except KeyError as err:
2700 raise exc.ArgumentError(
2701 "This text() construct doesn't define a "
2702 "bound parameter named %r" % bind._orig_key
2703 ) from err
2704 else:
2705 new_params[existing._orig_key] = bind
2707 for key, value in names_to_values.items():
2708 try:
2709 existing = new_params[key]
2710 except KeyError as err:
2711 raise exc.ArgumentError(
2712 "This text() construct doesn't define a "
2713 "bound parameter named %r" % key
2714 ) from err
2715 else:
2716 new_params[key] = existing._with_value(value, required=False)
2717 return self
2719 @property
2720 def type(self) -> TypeEngine[Any]:
2721 return type_api.NULLTYPE
2723 @property
2724 def comparator(self):
2725 # TODO: this seems wrong, it seems like we might not
2726 # be using this method.
2727 return self.type.comparator_factory(self) # type: ignore
2729 def self_group(
2730 self, against: Optional[OperatorType] = None
2731 ) -> Union[Self, Grouping[Any]]:
2732 if against is operators.in_op:
2733 return Grouping(self)
2734 else:
2735 return self
2738class TString(AbstractTextClause, inspection.Inspectable["TString"]):
2739 """Represent a SQL template string using Python 3.14+ t-strings.
2741 E.g.::
2743 from sqlalchemy import tstring, column
2745 a = 5
2746 b = 10
2747 stmt = tstring(t"select {a}, {b}")
2748 result = connection.execute(stmt)
2750 The :class:`_expression.TString` construct is produced using the
2751 :func:`_expression.tstring` function; see that function for full
2752 documentation.
2754 .. versionadded:: 2.1
2756 .. seealso::
2758 :func:`_expression.tstring`
2760 """
2762 __visit_name__ = "tstring"
2764 _traverse_internals: _TraverseInternalsType = [
2765 ("parts", InternalTraversal.dp_clauseelement_list)
2766 ] + ExecutableStatement._executable_traverse_internals
2768 @property
2769 def _is_star(self) -> bool: # type: ignore[override]
2770 return (
2771 len(self.parts) == 1
2772 and isinstance(self.parts[0], TextClause)
2773 and self.parts[0]._is_star
2774 )
2776 def __init__(self, template: Template):
2777 """Construct a :class:`_expression.TString` from a Python 3.14+
2778 template string.
2780 :param template: a Python 3.14+ template string (t-string) that
2781 contains SQL fragments and Python expressions to be interpolated.
2783 """
2784 self.parts: List[ClauseElement] = []
2786 if not isinstance(template, Template):
2787 raise exc.ArgumentError("pep-750 Tstring (e.g. t'...') expected")
2789 for part in template:
2790 if isinstance(part, str):
2791 self.parts.append(TextClause(part))
2792 else:
2793 assert hasattr(part, "value")
2794 self.parts.append(
2795 coercions.expect(roles.TStringElementRole, part.value)
2796 )
2798 def bindparams(
2799 self,
2800 *binds: BindParameter[Any],
2801 **names_to_values: Any,
2802 ) -> Self:
2803 """Not supported for TString constructs.
2805 TString constructs do not support .bindparams(). Bind parameters
2806 are automatically created from interpolated values.
2808 """
2809 raise NotImplementedError(
2810 "TString constructs do not support .bindparams(). "
2811 "Bind parameters are automatically created "
2812 "from interpolated values."
2813 )
2816class Null(SingletonConstant, roles.ConstExprRole[None], ColumnElement[None]):
2817 """Represent the NULL keyword in a SQL statement.
2819 :class:`.Null` is accessed as a constant via the
2820 :func:`.null` function.
2822 """
2824 __visit_name__ = "null"
2826 _traverse_internals: _TraverseInternalsType = []
2827 _singleton: Null
2829 if not TYPE_CHECKING:
2831 @util.memoized_property
2832 def type(self) -> TypeEngine[_T]: # noqa: A001
2833 return type_api.NULLTYPE
2835 @classmethod
2836 def _instance(cls) -> Null:
2837 """Return a constant :class:`.Null` construct."""
2839 return Null._singleton
2842Null._create_singleton()
2845class False_(
2846 SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]
2847):
2848 """Represent the ``false`` keyword, or equivalent, in a SQL statement.
2850 :class:`.False_` is accessed as a constant via the
2851 :func:`.false` function.
2853 """
2855 __visit_name__ = "false"
2856 _traverse_internals: _TraverseInternalsType = []
2857 _singleton: False_
2859 if not TYPE_CHECKING:
2861 @util.memoized_property
2862 def type(self) -> TypeEngine[_T]: # noqa: A001
2863 return type_api.BOOLEANTYPE
2865 def _negate(self) -> True_:
2866 return True_._singleton
2868 @classmethod
2869 def _instance(cls) -> False_:
2870 return False_._singleton
2873False_._create_singleton()
2876class True_(SingletonConstant, roles.ConstExprRole[bool], ColumnElement[bool]):
2877 """Represent the ``true`` keyword, or equivalent, in a SQL statement.
2879 :class:`.True_` is accessed as a constant via the
2880 :func:`.true` function.
2882 """
2884 __visit_name__ = "true"
2886 _traverse_internals: _TraverseInternalsType = []
2887 _singleton: True_
2889 if not TYPE_CHECKING:
2891 @util.memoized_property
2892 def type(self) -> TypeEngine[_T]: # noqa: A001
2893 return type_api.BOOLEANTYPE
2895 def _negate(self) -> False_:
2896 return False_._singleton
2898 @classmethod
2899 def _ifnone(
2900 cls, other: Optional[ColumnElement[Any]]
2901 ) -> ColumnElement[Any]:
2902 if other is None:
2903 return cls._instance()
2904 else:
2905 return other
2907 @classmethod
2908 def _instance(cls) -> True_:
2909 return True_._singleton
2912True_._create_singleton()
2915class ElementList(DQLDMLClauseElement):
2916 """Describe a list of clauses that will be space separated.
2918 This is a minimal version of :class:`.ClauseList` which is used by
2919 the :class:`.HasSyntaxExtension` class. It does not do any coercions
2920 so should be used internally only.
2922 .. versionadded:: 2.1
2924 """
2926 __visit_name__ = "element_list"
2928 _traverse_internals: _TraverseInternalsType = [
2929 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2930 ]
2932 clauses: typing_Tuple[ClauseElement, ...]
2934 def __init__(self, clauses: Sequence[ClauseElement]):
2935 self.clauses = tuple(clauses)
2938class OrderByList(
2939 roles.OrderByRole,
2940 operators.OrderingOperators,
2941 DQLDMLClauseElement,
2942):
2943 """Describe a list of clauses that will be comma separated to nest
2944 within an ORDER BY.
2946 .. versionadded:: 2.1
2948 """
2950 __visit_name__ = "order_by_list"
2952 _traverse_internals: _TraverseInternalsType = [
2953 ("clauses", InternalTraversal.dp_clauseelement_tuple),
2954 ]
2956 clauses: List[ColumnElement[Any]]
2958 def __init__(
2959 self,
2960 clauses: Iterable[Union[OrderByList, _ColumnExpressionArgument[Any]]],
2961 ):
2962 text_converter_role: Type[roles.SQLRole] = roles.ByOfRole
2963 self._text_converter_role = text_converter_role
2965 self.clauses = [
2966 coercions.expect(
2967 text_converter_role, clause, apply_propagate_attrs=self
2968 )
2969 for clause in clauses
2970 ]
2972 def __iter__(self) -> Iterator[ColumnElement[Any]]:
2973 return iter(self.clauses)
2975 def __len__(self) -> int:
2976 return len(self.clauses)
2978 @property
2979 def _select_iterable(self) -> _SelectIterable:
2980 return itertools.chain.from_iterable(
2981 [elem._select_iterable for elem in self.clauses]
2982 )
2984 @util.ro_non_memoized_property
2985 def _from_objects(self) -> List[FromClause]:
2986 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
2988 def self_group(
2989 self, against: Optional[OperatorType] = None
2990 ) -> Union[Self, Grouping[Any]]:
2991 return self
2993 def desc(self) -> OrderByList:
2994 return OrderByList([e.desc() for e in self.clauses])
2996 def asc(self) -> OrderByList:
2997 return OrderByList([e.asc() for e in self.clauses])
2999 def nulls_first(self) -> OrderByList:
3000 return OrderByList([e.nulls_first() for e in self.clauses])
3002 def nulls_last(self) -> OrderByList:
3003 return OrderByList([e.nulls_last() for e in self.clauses])
3006class ClauseList(
3007 roles.InElementRole,
3008 roles.OrderByRole,
3009 roles.ColumnsClauseRole,
3010 roles.DMLColumnRole,
3011 DQLDMLClauseElement,
3012):
3013 """Describe a list of clauses, separated by an operator.
3015 By default, is comma-separated, such as a column listing.
3017 """
3019 __visit_name__ = "clauselist"
3021 # Used by ORM context.py to identify ClauseList objects in legacy
3022 # composite attribute queries (see test_query_cols_legacy test)
3023 _is_clause_list = True
3025 _traverse_internals: _TraverseInternalsType = [
3026 ("clauses", InternalTraversal.dp_clauseelement_list),
3027 ("operator", InternalTraversal.dp_operator),
3028 ]
3030 clauses: List[ColumnElement[Any]]
3032 def __init__(
3033 self,
3034 *clauses: _ColumnExpressionArgument[Any],
3035 operator: OperatorType = operators.comma_op,
3036 group: bool = True,
3037 group_contents: bool = True,
3038 _literal_as_text_role: Type[roles.SQLRole] = roles.WhereHavingRole,
3039 ):
3040 self.operator = operator
3041 self.group = group
3042 self.group_contents = group_contents
3043 clauses_iterator: Iterable[_ColumnExpressionArgument[Any]] = clauses
3044 text_converter_role: Type[roles.SQLRole] = _literal_as_text_role
3045 self._text_converter_role = text_converter_role
3047 if self.group_contents:
3048 self.clauses = [
3049 coercions.expect(
3050 text_converter_role, clause, apply_propagate_attrs=self
3051 ).self_group(against=self.operator)
3052 for clause in clauses_iterator
3053 ]
3054 else:
3055 self.clauses = [
3056 coercions.expect(
3057 text_converter_role, clause, apply_propagate_attrs=self
3058 )
3059 for clause in clauses_iterator
3060 ]
3061 self._is_implicitly_boolean = operators.is_boolean(self.operator)
3063 @classmethod
3064 def _construct_raw(
3065 cls,
3066 operator: OperatorType,
3067 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
3068 ) -> ClauseList:
3069 self = cls.__new__(cls)
3070 self.clauses = list(clauses) if clauses else []
3071 self.group = True
3072 self.operator = operator
3073 self.group_contents = True
3074 self._is_implicitly_boolean = False
3075 return self
3077 def __iter__(self) -> Iterator[ColumnElement[Any]]:
3078 return iter(self.clauses)
3080 def __len__(self) -> int:
3081 return len(self.clauses)
3083 @property
3084 def _select_iterable(self) -> _SelectIterable:
3085 return itertools.chain.from_iterable(
3086 [elem._select_iterable for elem in self.clauses]
3087 )
3089 def append(self, clause):
3090 if self.group_contents:
3091 self.clauses.append(
3092 coercions.expect(self._text_converter_role, clause).self_group(
3093 against=self.operator
3094 )
3095 )
3096 else:
3097 self.clauses.append(
3098 coercions.expect(self._text_converter_role, clause)
3099 )
3101 @util.ro_non_memoized_property
3102 def _from_objects(self) -> List[FromClause]:
3103 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
3105 def self_group(
3106 self, against: Optional[OperatorType] = None
3107 ) -> Union[Self, Grouping[Any]]:
3108 if self.group and operators.is_precedent(self.operator, against):
3109 return Grouping(self)
3110 else:
3111 return self
3114class OperatorExpression(ColumnElement[_T]):
3115 """base for expressions that contain an operator and operands
3117 .. versionadded:: 2.0
3119 """
3121 operator: OperatorType
3122 type: TypeEngine[_T]
3124 group: bool = True
3126 @property
3127 def is_comparison(self):
3128 return operators.is_comparison(self.operator)
3130 def self_group(
3131 self, against: Optional[OperatorType] = None
3132 ) -> Union[Self, Grouping[_T]]:
3133 if (
3134 self.group
3135 and operators.is_precedent(self.operator, against)
3136 or (
3137 # a negate against a non-boolean operator
3138 # doesn't make too much sense but we should
3139 # group for that
3140 against is operators.inv
3141 and not operators.is_boolean(self.operator)
3142 )
3143 ):
3144 return Grouping(self)
3145 else:
3146 return self
3148 @property
3149 def _flattened_operator_clauses(
3150 self,
3151 ) -> typing_Tuple[ColumnElement[Any], ...]:
3152 raise NotImplementedError()
3154 @classmethod
3155 def _construct_for_op(
3156 cls,
3157 left: ColumnElement[Any],
3158 right: ColumnElement[Any],
3159 op: OperatorType,
3160 *,
3161 type_: TypeEngine[_T],
3162 negate: Optional[OperatorType] = None,
3163 modifiers: Optional[Mapping[str, Any]] = None,
3164 ) -> OperatorExpression[_T]:
3165 if operators.is_associative(op):
3166 assert (
3167 negate is None
3168 ), f"negate not supported for associative operator {op}"
3170 multi = False
3171 if getattr(
3172 left, "operator", None
3173 ) is op and type_._compare_type_affinity(left.type):
3174 multi = True
3175 left_flattened = left._flattened_operator_clauses
3176 else:
3177 left_flattened = (left,)
3179 if getattr(
3180 right, "operator", None
3181 ) is op and type_._compare_type_affinity(right.type):
3182 multi = True
3183 right_flattened = right._flattened_operator_clauses
3184 else:
3185 right_flattened = (right,)
3187 if multi:
3188 return ExpressionClauseList._construct_for_list(
3189 op,
3190 type_,
3191 *(left_flattened + right_flattened),
3192 )
3194 if right._is_collection_aggregate:
3195 negate = None
3197 return BinaryExpression(
3198 left, right, op, type_=type_, negate=negate, modifiers=modifiers
3199 )
3202class ExpressionClauseList(OperatorExpression[_T]):
3203 """Describe a list of clauses, separated by an operator,
3204 in a column expression context.
3206 :class:`.ExpressionClauseList` differs from :class:`.ClauseList` in that
3207 it represents a column-oriented DQL expression only, not an open ended
3208 list of anything comma separated.
3210 .. versionadded:: 2.0
3212 """
3214 __visit_name__ = "expression_clauselist"
3216 _traverse_internals: _TraverseInternalsType = [
3217 ("clauses", InternalTraversal.dp_clauseelement_tuple),
3218 ("operator", InternalTraversal.dp_operator),
3219 ]
3221 clauses: typing_Tuple[ColumnElement[Any], ...]
3223 group: bool
3225 def __init__(
3226 self,
3227 operator: OperatorType,
3228 *clauses: _ColumnExpressionArgument[Any],
3229 type_: Optional[_TypeEngineArgument[_T]] = None,
3230 ):
3231 self.operator = operator
3233 self.clauses = tuple(
3234 coercions.expect(
3235 roles.ExpressionElementRole, clause, apply_propagate_attrs=self
3236 )
3237 for clause in clauses
3238 )
3239 self._is_implicitly_boolean = operators.is_boolean(self.operator)
3240 self.type = type_api.to_instance(type_) # type: ignore
3242 @property
3243 def _flattened_operator_clauses(
3244 self,
3245 ) -> typing_Tuple[ColumnElement[Any], ...]:
3246 return self.clauses
3248 def __iter__(self) -> Iterator[ColumnElement[Any]]:
3249 return iter(self.clauses)
3251 def __len__(self) -> int:
3252 return len(self.clauses)
3254 @property
3255 def _select_iterable(self) -> _SelectIterable:
3256 return (self,)
3258 @util.ro_non_memoized_property
3259 def _from_objects(self) -> List[FromClause]:
3260 return list(itertools.chain(*[c._from_objects for c in self.clauses]))
3262 def _append_inplace(self, clause: ColumnElement[Any]) -> None:
3263 self.clauses += (clause,)
3265 @classmethod
3266 def _construct_for_list(
3267 cls,
3268 operator: OperatorType,
3269 type_: TypeEngine[_T],
3270 *clauses: ColumnElement[Any],
3271 group: bool = True,
3272 ) -> ExpressionClauseList[_T]:
3273 self = cls.__new__(cls)
3274 self.group = group
3275 if group:
3276 self.clauses = tuple(
3277 c.self_group(against=operator) for c in clauses
3278 )
3279 else:
3280 self.clauses = clauses
3281 self.operator = operator
3282 self.type = type_
3283 for c in clauses:
3284 if c._propagate_attrs:
3285 self._propagate_attrs = c._propagate_attrs
3286 break
3287 return self
3289 def _negate(self) -> Any:
3290 grouped = self.self_group(against=operators.inv)
3291 assert isinstance(grouped, ColumnElement)
3292 return UnaryExpression(grouped, operator=operators.inv)
3295class BooleanClauseList(ExpressionClauseList[bool]):
3296 __visit_name__ = "expression_clauselist"
3297 inherit_cache = True
3299 def __init__(self, *arg, **kw):
3300 raise NotImplementedError(
3301 "BooleanClauseList has a private constructor"
3302 )
3304 @classmethod
3305 def _process_clauses_for_boolean(
3306 cls,
3307 operator: OperatorType,
3308 continue_on: Any,
3309 skip_on: Any,
3310 clauses: Iterable[ColumnElement[Any]],
3311 ) -> typing_Tuple[int, List[ColumnElement[Any]]]:
3312 has_continue_on = None
3314 convert_clauses = []
3316 against = operators._asbool
3317 lcc = 0
3319 for clause in clauses:
3320 if clause is continue_on:
3321 # instance of continue_on, like and_(x, y, True, z), store it
3322 # if we didn't find one already, we will use it if there
3323 # are no other expressions here.
3324 has_continue_on = clause
3325 elif clause is skip_on:
3326 # instance of skip_on, e.g. and_(x, y, False, z), cancels
3327 # the rest out
3328 convert_clauses = [clause]
3329 lcc = 1
3330 break
3331 else:
3332 if not lcc:
3333 lcc = 1
3334 else:
3335 against = operator
3336 # technically this would be len(convert_clauses) + 1
3337 # however this only needs to indicate "greater than one"
3338 lcc = 2
3339 convert_clauses.append(clause)
3341 if not convert_clauses and has_continue_on is not None:
3342 convert_clauses = [has_continue_on]
3343 lcc = 1
3345 return lcc, [c.self_group(against=against) for c in convert_clauses]
3347 @classmethod
3348 def _construct(
3349 cls,
3350 operator: OperatorType,
3351 continue_on: Any,
3352 skip_on: Any,
3353 initial_clause: Any = _NoArg.NO_ARG,
3354 *clauses: Any,
3355 **kw: Any,
3356 ) -> ColumnElement[Any]:
3357 if initial_clause is _NoArg.NO_ARG:
3358 # no elements period. deprecated use case. return an empty
3359 # ClauseList construct that generates nothing unless it has
3360 # elements added to it.
3361 name = operator.__name__
3363 util.warn_deprecated(
3364 f"Invoking {name}() without arguments is deprecated, and "
3365 f"will be disallowed in a future release. For an empty "
3366 f"""{name}() construct, use '{name}({
3367 'true()' if continue_on is True_._singleton else 'false()'
3368 }, *args)' """
3369 f"""or '{name}({
3370 'True' if continue_on is True_._singleton else 'False'
3371 }, *args)'.""",
3372 version="1.4",
3373 )
3374 return cls._construct_raw(operator)
3376 lcc, convert_clauses = cls._process_clauses_for_boolean(
3377 operator,
3378 continue_on,
3379 skip_on,
3380 [
3381 coercions.expect(roles.WhereHavingRole, clause)
3382 for clause in util.coerce_generator_arg(
3383 (initial_clause,) + clauses
3384 )
3385 ],
3386 )
3388 if lcc > 1:
3389 # multiple elements. Return regular BooleanClauseList
3390 # which will link elements against the operator.
3392 flattened_clauses = itertools.chain.from_iterable(
3393 (
3394 (c for c in to_flat._flattened_operator_clauses)
3395 if getattr(to_flat, "operator", None) is operator
3396 else (to_flat,)
3397 )
3398 for to_flat in convert_clauses
3399 )
3401 return cls._construct_raw(operator, flattened_clauses) # type: ignore # noqa: E501
3402 else:
3403 assert lcc
3404 # just one element. return it as a single boolean element,
3405 # not a list and discard the operator.
3406 return convert_clauses[0]
3408 @classmethod
3409 def _construct_for_whereclause(
3410 cls, clauses: Iterable[ColumnElement[Any]]
3411 ) -> Optional[ColumnElement[bool]]:
3412 operator, continue_on, skip_on = (
3413 operators.and_,
3414 True_._singleton,
3415 False_._singleton,
3416 )
3418 lcc, convert_clauses = cls._process_clauses_for_boolean(
3419 operator,
3420 continue_on,
3421 skip_on,
3422 clauses, # these are assumed to be coerced already
3423 )
3425 if lcc > 1:
3426 # multiple elements. Return regular BooleanClauseList
3427 # which will link elements against the operator.
3428 return cls._construct_raw(operator, convert_clauses)
3429 elif lcc == 1:
3430 # just one element. return it as a single boolean element,
3431 # not a list and discard the operator.
3432 return convert_clauses[0]
3433 else:
3434 return None
3436 @classmethod
3437 def _construct_raw(
3438 cls,
3439 operator: OperatorType,
3440 clauses: Optional[Sequence[ColumnElement[Any]]] = None,
3441 ) -> BooleanClauseList:
3442 self = cls.__new__(cls)
3443 self.clauses = tuple(clauses) if clauses else ()
3444 self.group = True
3445 self.operator = operator
3446 self.type = type_api.BOOLEANTYPE
3447 self._is_implicitly_boolean = True
3448 return self
3450 @classmethod
3451 def and_(
3452 cls,
3453 initial_clause: Union[
3454 Literal[True], _ColumnExpressionArgument[bool], _NoArg
3455 ] = _NoArg.NO_ARG,
3456 *clauses: _ColumnExpressionArgument[bool],
3457 ) -> ColumnElement[bool]:
3458 r"""Produce a conjunction of expressions joined by ``AND``.
3460 See :func:`_sql.and_` for full documentation.
3461 """
3462 return cls._construct(
3463 operators.and_,
3464 True_._singleton,
3465 False_._singleton,
3466 initial_clause,
3467 *clauses,
3468 )
3470 @classmethod
3471 def or_(
3472 cls,
3473 initial_clause: Union[
3474 Literal[False], _ColumnExpressionArgument[bool], _NoArg
3475 ] = _NoArg.NO_ARG,
3476 *clauses: _ColumnExpressionArgument[bool],
3477 ) -> ColumnElement[bool]:
3478 """Produce a conjunction of expressions joined by ``OR``.
3480 See :func:`_sql.or_` for full documentation.
3481 """
3482 return cls._construct(
3483 operators.or_,
3484 False_._singleton,
3485 True_._singleton,
3486 initial_clause,
3487 *clauses,
3488 )
3490 @property
3491 def _select_iterable(self) -> _SelectIterable:
3492 return (self,)
3494 def self_group(
3495 self, against: Optional[OperatorType] = None
3496 ) -> Union[Self, Grouping[bool]]:
3497 if not self.clauses:
3498 return self
3499 else:
3500 return super().self_group(against=against)
3503and_ = BooleanClauseList.and_
3504or_ = BooleanClauseList.or_
3507class Tuple(ClauseList, ColumnElement[TupleAny]):
3508 """Represent a SQL tuple."""
3510 __visit_name__ = "tuple"
3512 _traverse_internals: _TraverseInternalsType = (
3513 ClauseList._traverse_internals + []
3514 )
3516 type: TupleType
3518 @util.preload_module("sqlalchemy.sql.sqltypes")
3519 def __init__(
3520 self,
3521 *clauses: _ColumnExpressionArgument[Any],
3522 types: Optional[Sequence[_TypeEngineArgument[Any]]] = None,
3523 ):
3524 sqltypes = util.preloaded.sql_sqltypes
3526 if types is None:
3527 init_clauses: List[ColumnElement[Any]] = [
3528 coercions.expect(roles.ExpressionElementRole, c)
3529 for c in clauses
3530 ]
3531 else:
3532 if len(types) != len(clauses):
3533 raise exc.ArgumentError(
3534 "Wrong number of elements for %d-tuple: %r "
3535 % (len(types), clauses)
3536 )
3537 init_clauses = [
3538 coercions.expect(
3539 roles.ExpressionElementRole,
3540 c,
3541 type_=typ if not typ._isnull else None,
3542 )
3543 for typ, c in zip(types, clauses)
3544 ]
3546 self.type = sqltypes.TupleType(*[arg.type for arg in init_clauses])
3547 super().__init__(*init_clauses)
3549 @property
3550 def _select_iterable(self) -> _SelectIterable:
3551 return (self,)
3553 def _bind_param(self, operator, obj, type_=None, expanding=False):
3554 if expanding:
3555 return BindParameter(
3556 None,
3557 value=obj,
3558 _compared_to_operator=operator,
3559 unique=True,
3560 expanding=True,
3561 type_=type_,
3562 _compared_to_type=self.type,
3563 )
3564 else:
3565 return Tuple(
3566 *[
3567 BindParameter(
3568 None,
3569 o,
3570 _compared_to_operator=operator,
3571 _compared_to_type=compared_to_type,
3572 unique=True,
3573 type_=type_,
3574 )
3575 for o, compared_to_type in zip(obj, self.type.types)
3576 ]
3577 )
3579 def self_group(self, against: Optional[OperatorType] = None) -> Self:
3580 # Tuple is parenthesized by definition.
3581 return self
3584class Case(ColumnElement[_T]):
3585 """Represent a ``CASE`` expression.
3587 :class:`.Case` is produced using the :func:`.case` factory function,
3588 as in::
3590 from sqlalchemy import case
3592 stmt = select(users_table).where(
3593 case(
3594 (users_table.c.name == "wendy", "W"),
3595 (users_table.c.name == "jack", "J"),
3596 else_="E",
3597 )
3598 )
3600 Details on :class:`.Case` usage is at :func:`.case`.
3602 .. seealso::
3604 :func:`.case`
3606 """
3608 __visit_name__ = "case"
3610 _traverse_internals: _TraverseInternalsType = [
3611 ("value", InternalTraversal.dp_clauseelement),
3612 ("whens", InternalTraversal.dp_clauseelement_tuples),
3613 ("else_", InternalTraversal.dp_clauseelement),
3614 ]
3616 # for case(), the type is derived from the whens. so for the moment
3617 # users would have to cast() the case to get a specific type
3619 whens: List[typing_Tuple[ColumnElement[bool], ColumnElement[_T]]]
3620 else_: Optional[ColumnElement[_T]]
3621 value: Optional[ColumnElement[Any]]
3623 def __init__(
3624 self,
3625 *whens: Union[
3626 typing_Tuple[_ColumnExpressionArgument[bool], Any],
3627 Mapping[Any, Any],
3628 ],
3629 value: Optional[Any] = None,
3630 else_: Optional[Any] = None,
3631 ):
3632 new_whens: Iterable[Any] = coercions._expression_collection_was_a_list(
3633 "whens", "case", whens
3634 )
3635 try:
3636 new_whens = util.dictlike_iteritems(new_whens)
3637 except TypeError:
3638 pass
3640 self.whens = [
3641 (
3642 coercions.expect(
3643 roles.ExpressionElementRole,
3644 c,
3645 apply_propagate_attrs=self,
3646 ).self_group(),
3647 coercions.expect(roles.ExpressionElementRole, r),
3648 )
3649 for (c, r) in new_whens
3650 ]
3652 if value is None:
3653 self.value = None
3654 else:
3655 self.value = coercions.expect(roles.ExpressionElementRole, value)
3657 if else_ is not None:
3658 self.else_ = coercions.expect(roles.ExpressionElementRole, else_)
3659 else:
3660 self.else_ = None
3662 type_ = next(
3663 (
3664 then.type
3665 # Iterate `whens` in reverse to match previous behaviour
3666 # where type of final element took priority
3667 for *_, then in reversed(self.whens)
3668 if not then.type._isnull
3669 ),
3670 self.else_.type if self.else_ is not None else type_api.NULLTYPE,
3671 )
3672 self.type = cast(_T, type_)
3674 @util.ro_non_memoized_property
3675 def _from_objects(self) -> List[FromClause]:
3676 return list(
3677 itertools.chain(*[x._from_objects for x in self.get_children()])
3678 )
3681class Cast(WrapsColumnExpression[_T]):
3682 """Represent a ``CAST`` expression.
3684 :class:`.Cast` is produced using the :func:`.cast` factory function,
3685 as in::
3687 from sqlalchemy import cast, Numeric
3689 stmt = select(cast(product_table.c.unit_price, Numeric(10, 4)))
3691 Details on :class:`.Cast` usage is at :func:`.cast`.
3693 .. seealso::
3695 :ref:`tutorial_casts`
3697 :func:`.cast`
3699 :func:`.try_cast`
3701 :func:`.type_coerce` - an alternative to CAST that coerces the type
3702 on the Python side only, which is often sufficient to generate the
3703 correct SQL and data coercion.
3705 """
3707 __visit_name__ = "cast"
3709 _traverse_internals: _TraverseInternalsType = [
3710 ("clause", InternalTraversal.dp_clauseelement),
3711 ("type", InternalTraversal.dp_type),
3712 ]
3714 clause: ColumnElement[Any]
3715 type: TypeEngine[_T]
3716 typeclause: TypeClause
3718 def __init__(
3719 self,
3720 expression: _ColumnExpressionArgument[Any],
3721 type_: _TypeEngineArgument[_T],
3722 ):
3723 self.type = type_api.to_instance(type_)
3724 self.clause = coercions.expect(
3725 roles.ExpressionElementRole,
3726 expression,
3727 type_=self.type,
3728 apply_propagate_attrs=self,
3729 )
3730 self.typeclause = TypeClause(self.type)
3732 @util.ro_non_memoized_property
3733 def _from_objects(self) -> List[FromClause]:
3734 return self.clause._from_objects
3736 @property
3737 def wrapped_column_expression(self):
3738 return self.clause
3741class TryCast(Cast[_T]):
3742 """Represent a TRY_CAST expression.
3744 Details on :class:`.TryCast` usage is at :func:`.try_cast`.
3746 .. seealso::
3748 :func:`.try_cast`
3750 :ref:`tutorial_casts`
3751 """
3753 __visit_name__ = "try_cast"
3754 inherit_cache = True
3757class TypeCoerce(WrapsColumnExpression[_T]):
3758 """Represent a Python-side type-coercion wrapper.
3760 :class:`.TypeCoerce` supplies the :func:`_expression.type_coerce`
3761 function; see that function for usage details.
3763 .. seealso::
3765 :func:`_expression.type_coerce`
3767 :func:`.cast`
3769 """
3771 __visit_name__ = "type_coerce"
3773 _traverse_internals: _TraverseInternalsType = [
3774 ("clause", InternalTraversal.dp_clauseelement),
3775 ("type", InternalTraversal.dp_type),
3776 ]
3778 clause: ColumnElement[Any]
3779 type: TypeEngine[_T]
3781 def __init__(
3782 self,
3783 expression: _ColumnExpressionArgument[Any],
3784 type_: _TypeEngineArgument[_T],
3785 ):
3786 self.type = type_api.to_instance(type_)
3787 self.clause = coercions.expect(
3788 roles.ExpressionElementRole,
3789 expression,
3790 type_=self.type,
3791 apply_propagate_attrs=self,
3792 )
3794 @util.ro_non_memoized_property
3795 def _from_objects(self) -> List[FromClause]:
3796 return self.clause._from_objects
3798 @HasMemoized.memoized_attribute
3799 def typed_expression(self):
3800 if isinstance(self.clause, BindParameter):
3801 bp = self.clause._clone()
3802 bp.type = self.type
3803 return bp
3804 else:
3805 return self.clause
3807 @property
3808 def wrapped_column_expression(self):
3809 return self.clause
3811 def self_group(
3812 self, against: Optional[OperatorType] = None
3813 ) -> TypeCoerce[_T]:
3814 grouped = self.clause.self_group(against=against)
3815 if grouped is not self.clause:
3816 return TypeCoerce(grouped, self.type)
3817 else:
3818 return self
3821class Extract(ColumnElement[int]):
3822 """Represent a SQL EXTRACT clause, ``extract(field FROM expr)``."""
3824 __visit_name__ = "extract"
3826 _traverse_internals: _TraverseInternalsType = [
3827 ("expr", InternalTraversal.dp_clauseelement),
3828 ("field", InternalTraversal.dp_string),
3829 ]
3831 expr: ColumnElement[Any]
3832 field: str
3834 def __init__(self, field: str, expr: _ColumnExpressionArgument[Any]):
3835 self.type = type_api.INTEGERTYPE
3836 self.field = field
3837 self.expr = coercions.expect(roles.ExpressionElementRole, expr)
3839 @util.ro_non_memoized_property
3840 def _from_objects(self) -> List[FromClause]:
3841 return self.expr._from_objects
3844class _label_reference(ColumnElement[_T]):
3845 """Wrap a column expression as it appears in a 'reference' context.
3847 This expression is any that includes an _order_by_label_element,
3848 which is a Label, or a DESC / ASC construct wrapping a Label.
3850 The production of _label_reference() should occur when an expression
3851 is added to this context; this includes the ORDER BY or GROUP BY of a
3852 SELECT statement, as well as a few other places, such as the ORDER BY
3853 within an OVER clause.
3855 """
3857 __visit_name__ = "label_reference"
3859 _traverse_internals: _TraverseInternalsType = [
3860 ("element", InternalTraversal.dp_clauseelement)
3861 ]
3863 element: ColumnElement[_T]
3865 def __init__(self, element: ColumnElement[_T]):
3866 self.element = element
3867 self._propagate_attrs = element._propagate_attrs
3869 @util.ro_non_memoized_property
3870 def _from_objects(self) -> List[FromClause]:
3871 return []
3874class _textual_label_reference(ColumnElement[Any]):
3875 __visit_name__ = "textual_label_reference"
3877 _traverse_internals: _TraverseInternalsType = [
3878 ("element", InternalTraversal.dp_string)
3879 ]
3881 def __init__(self, element: str):
3882 self.element = element
3884 @util.memoized_property
3885 def _text_clause(self) -> TextClause:
3886 return TextClause(self.element)
3889class UnaryExpression(ColumnElement[_T]):
3890 """Define a 'unary' expression.
3892 A unary expression has a single column expression
3893 and an operator. The operator can be placed on the left
3894 (where it is called the 'operator') or right (where it is called the
3895 'modifier') of the column expression.
3897 :class:`.UnaryExpression` is the basis for several unary operators
3898 including those used by :func:`.desc`, :func:`.asc`, :func:`.distinct`,
3899 :func:`.nulls_first` and :func:`.nulls_last`.
3901 """
3903 __visit_name__ = "unary"
3905 _traverse_internals: _TraverseInternalsType = [
3906 ("element", InternalTraversal.dp_clauseelement),
3907 ("operator", InternalTraversal.dp_operator),
3908 ("modifier", InternalTraversal.dp_operator),
3909 ]
3911 element: ColumnElement[Any]
3912 operator: Optional[OperatorType]
3913 modifier: Optional[OperatorType]
3915 def __init__(
3916 self,
3917 element: ColumnElement[Any],
3918 *,
3919 operator: Optional[OperatorType] = None,
3920 modifier: Optional[OperatorType] = None,
3921 type_: Optional[_TypeEngineArgument[_T]] = None,
3922 wraps_column_expression: bool = False, # legacy, not used as of 2.0.42
3923 ):
3924 self.operator = operator
3925 self.modifier = modifier
3926 self._propagate_attrs = element._propagate_attrs
3927 self.element = element.self_group(
3928 against=self.operator or self.modifier
3929 )
3931 # if type is None, we get NULLTYPE, which is our _T. But I don't
3932 # know how to get the overloads to express that correctly
3933 self.type = type_api.to_instance(type_) # type: ignore
3935 def _wraps_unnamed_column(self):
3936 ungrouped = self.element._ungroup()
3937 return (
3938 not isinstance(ungrouped, NamedColumn)
3939 or ungrouped._non_anon_label is None
3940 )
3942 @classmethod
3943 def _create_nulls_first(
3944 cls,
3945 column: _ColumnExpressionArgument[_T],
3946 ) -> UnaryExpression[_T]:
3947 return UnaryExpression(
3948 coercions.expect(roles.ByOfRole, column),
3949 modifier=operators.nulls_first_op,
3950 )
3952 @classmethod
3953 def _create_nulls_last(
3954 cls,
3955 column: _ColumnExpressionArgument[_T],
3956 ) -> UnaryExpression[_T]:
3957 return UnaryExpression(
3958 coercions.expect(roles.ByOfRole, column),
3959 modifier=operators.nulls_last_op,
3960 )
3962 @classmethod
3963 def _create_desc(
3964 cls, column: _ColumnExpressionOrStrLabelArgument[_T]
3965 ) -> UnaryExpression[_T]:
3967 return UnaryExpression(
3968 coercions.expect(roles.ByOfRole, column),
3969 modifier=operators.desc_op,
3970 )
3972 @classmethod
3973 def _create_asc(
3974 cls,
3975 column: _ColumnExpressionOrStrLabelArgument[_T],
3976 ) -> UnaryExpression[_T]:
3977 return UnaryExpression(
3978 coercions.expect(roles.ByOfRole, column),
3979 modifier=operators.asc_op,
3980 )
3982 @classmethod
3983 def _create_distinct(
3984 cls,
3985 expr: _ColumnExpressionArgument[_T],
3986 ) -> UnaryExpression[_T]:
3987 col_expr: ColumnElement[_T] = coercions.expect(
3988 roles.ExpressionElementRole, expr
3989 )
3990 return UnaryExpression(
3991 col_expr,
3992 operator=operators.distinct_op,
3993 type_=col_expr.type,
3994 )
3996 @classmethod
3997 def _create_bitwise_not(
3998 cls,
3999 expr: _ColumnExpressionArgument[_T],
4000 ) -> UnaryExpression[_T]:
4001 col_expr: ColumnElement[_T] = coercions.expect(
4002 roles.ExpressionElementRole, expr
4003 )
4004 return UnaryExpression(
4005 col_expr,
4006 operator=operators.bitwise_not_op,
4007 type_=col_expr.type,
4008 )
4010 @property
4011 def _order_by_label_element(self) -> Optional[Label[Any]]:
4012 if operators.is_order_by_modifier(self.modifier):
4013 return self.element._order_by_label_element
4014 else:
4015 return None
4017 @util.ro_non_memoized_property
4018 def _from_objects(self) -> List[FromClause]:
4019 return self.element._from_objects
4021 def _negate(self) -> ColumnElement[Any]:
4022 if self.type._type_affinity is type_api.BOOLEANTYPE._type_affinity:
4023 return UnaryExpression(
4024 self.self_group(against=operators.inv),
4025 operator=operators.inv,
4026 type_=type_api.BOOLEANTYPE,
4027 )
4028 else:
4029 return ColumnElement._negate(self)
4031 def self_group(
4032 self, against: Optional[OperatorType] = None
4033 ) -> Union[Self, Grouping[_T]]:
4034 if self.operator and operators.is_precedent(self.operator, against):
4035 return Grouping(self)
4036 else:
4037 return self
4040class CollectionAggregate(UnaryExpression[_T]):
4041 """Forms the basis for right-hand collection operator modifiers
4042 ANY and ALL.
4044 The ANY and ALL keywords are available in different ways on different
4045 backends. On PostgreSQL, they only work for an ARRAY type. On
4046 MySQL, they only work for subqueries.
4048 """
4050 inherit_cache = True
4051 _is_collection_aggregate = True
4053 @classmethod
4054 def _create_any(
4055 cls, expr: _ColumnExpressionArgument[_T]
4056 ) -> CollectionAggregate[bool]:
4057 """create CollectionAggregate for the legacy
4058 ARRAY.Comparator.any() method"""
4059 col_expr: ColumnElement[_T] = coercions.expect(
4060 roles.ExpressionElementRole,
4061 expr,
4062 )
4063 col_expr = col_expr.self_group()
4064 return CollectionAggregate(
4065 col_expr,
4066 operator=operators.any_op,
4067 type_=type_api.BOOLEANTYPE,
4068 )
4070 @classmethod
4071 def _create_all(
4072 cls, expr: _ColumnExpressionArgument[_T]
4073 ) -> CollectionAggregate[bool]:
4074 """create CollectionAggregate for the legacy
4075 ARRAY.Comparator.all() method"""
4076 col_expr: ColumnElement[_T] = coercions.expect(
4077 roles.ExpressionElementRole,
4078 expr,
4079 )
4080 col_expr = col_expr.self_group()
4081 return CollectionAggregate(
4082 col_expr,
4083 operator=operators.all_op,
4084 type_=type_api.BOOLEANTYPE,
4085 )
4087 @util.preload_module("sqlalchemy.sql.sqltypes")
4088 def _bind_param(
4089 self,
4090 operator: operators.OperatorType,
4091 obj: Any,
4092 type_: Optional[TypeEngine[_T]] = None,
4093 expanding: bool = False,
4094 ) -> BindParameter[_T]:
4095 """For new style any_(), all_(), ensure compared literal value
4096 receives appropriate bound parameter type."""
4098 # a CollectionAggregate is specific to ARRAY or int
4099 # only. So for ARRAY case, make sure we use correct element type
4100 sqltypes = util.preloaded.sql_sqltypes
4101 if self.element.type._type_affinity is sqltypes.ARRAY:
4102 compared_to_type = cast(
4103 sqltypes.ARRAY[Any], self.element.type
4104 ).item_type
4105 else:
4106 compared_to_type = self.element.type
4108 return BindParameter(
4109 None,
4110 obj,
4111 _compared_to_operator=operator,
4112 type_=type_,
4113 _compared_to_type=compared_to_type,
4114 unique=True,
4115 expanding=expanding,
4116 )
4118 # operate and reverse_operate are hardwired to
4119 # dispatch onto the type comparator directly, so that we can
4120 # ensure "reversed" behavior.
4121 def operate(
4122 self, op: OperatorType, *other: Any, **kwargs: Any
4123 ) -> ColumnElement[_T]:
4124 if not operators.is_comparison(op):
4125 raise exc.ArgumentError(
4126 "Only comparison operators may be used with ANY/ALL"
4127 )
4128 kwargs["reverse"] = True
4129 return self.comparator.operate(operators.mirror(op), *other, **kwargs)
4131 def reverse_operate(
4132 self, op: OperatorType, other: Any, **kwargs: Any
4133 ) -> ColumnElement[_T]:
4134 # comparison operators should never call reverse_operate
4135 assert not operators.is_comparison(op)
4136 raise exc.ArgumentError(
4137 "Only comparison operators may be used with ANY/ALL"
4138 )
4141class AsBoolean(WrapsColumnExpression[bool], UnaryExpression[bool]):
4142 inherit_cache = True
4144 def __init__(self, element, operator, negate):
4145 self.element = element
4146 self.type = type_api.BOOLEANTYPE
4147 self.operator = operator
4148 self.negate = negate
4149 self.modifier = None
4150 self._is_implicitly_boolean = element._is_implicitly_boolean
4152 @property
4153 def wrapped_column_expression(self):
4154 return self.element
4156 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4157 return self
4159 def _negate(self):
4160 if isinstance(self.element, (True_, False_)):
4161 return self.element._negate()
4162 else:
4163 return AsBoolean(self.element, self.negate, self.operator)
4166class BinaryExpression(OperatorExpression[_T]):
4167 """Represent an expression that is ``LEFT <operator> RIGHT``.
4169 A :class:`.BinaryExpression` is generated automatically
4170 whenever two column expressions are used in a Python binary expression:
4172 .. sourcecode:: pycon+sql
4174 >>> from sqlalchemy.sql import column
4175 >>> column("a") + column("b")
4176 <sqlalchemy.sql.expression.BinaryExpression object at 0x101029dd0>
4177 >>> print(column("a") + column("b"))
4178 {printsql}a + b
4180 """
4182 __visit_name__ = "binary"
4184 _traverse_internals: _TraverseInternalsType = [
4185 ("left", InternalTraversal.dp_clauseelement),
4186 ("right", InternalTraversal.dp_clauseelement),
4187 ("operator", InternalTraversal.dp_operator),
4188 ("negate", InternalTraversal.dp_operator),
4189 ("modifiers", InternalTraversal.dp_plain_dict),
4190 (
4191 "type",
4192 InternalTraversal.dp_type,
4193 ),
4194 ]
4196 _cache_key_traversal = [
4197 ("left", InternalTraversal.dp_clauseelement),
4198 ("right", InternalTraversal.dp_clauseelement),
4199 ("operator", InternalTraversal.dp_operator),
4200 ("modifiers", InternalTraversal.dp_plain_dict),
4201 # "type" affects JSON CAST operators, so while redundant in most cases,
4202 # is needed for that one
4203 (
4204 "type",
4205 InternalTraversal.dp_type,
4206 ),
4207 ]
4209 _is_implicitly_boolean = True
4210 """Indicates that any database will know this is a boolean expression
4211 even if the database does not have an explicit boolean datatype.
4213 """
4215 left: ColumnElement[Any]
4216 right: ColumnElement[Any]
4217 modifiers: Mapping[str, Any]
4219 def __init__(
4220 self,
4221 left: ColumnElement[Any],
4222 right: ColumnElement[Any],
4223 operator: OperatorType,
4224 type_: Optional[_TypeEngineArgument[_T]] = None,
4225 negate: Optional[OperatorType] = None,
4226 modifiers: Optional[Mapping[str, Any]] = None,
4227 ):
4228 # allow compatibility with libraries that
4229 # refer to BinaryExpression directly and pass strings
4230 if isinstance(operator, str):
4231 operator = operators.custom_op(operator)
4232 self._orig = (left.__hash__(), right.__hash__())
4233 self._propagate_attrs = left._propagate_attrs or right._propagate_attrs
4234 self.left = left.self_group(against=operator)
4235 self.right = right.self_group(against=operator)
4236 self.operator = operator
4238 # if type is None, we get NULLTYPE, which is our _T. But I don't
4239 # know how to get the overloads to express that correctly
4240 self.type = type_api.to_instance(type_) # type: ignore
4242 self.negate = negate
4243 self._is_implicitly_boolean = operators.is_boolean(operator)
4245 if modifiers is None:
4246 self.modifiers = {}
4247 else:
4248 self.modifiers = modifiers
4250 @property
4251 def _flattened_operator_clauses(
4252 self,
4253 ) -> typing_Tuple[ColumnElement[Any], ...]:
4254 return (self.left, self.right)
4256 def __bool__(self):
4257 """Implement Python-side "bool" for BinaryExpression as a
4258 simple "identity" check for the left and right attributes,
4259 if the operator is "eq" or "ne". Otherwise the expression
4260 continues to not support "bool" like all other column expressions.
4262 The rationale here is so that ColumnElement objects can be hashable.
4263 What? Well, suppose you do this::
4265 c1, c2 = column("x"), column("y")
4266 s1 = set([c1, c2])
4268 We do that **a lot**, columns inside of sets is an extremely basic
4269 thing all over the ORM for example.
4271 So what happens if we do this? ::
4273 c1 in s1
4275 Hashing means it will normally use ``__hash__()`` of the object,
4276 but in case of hash collision, it's going to also do ``c1 == c1``
4277 and/or ``c1 == c2`` inside. Those operations need to return a
4278 True/False value. But because we override ``==`` and ``!=``, they're
4279 going to get a BinaryExpression. Hence we implement ``__bool__`` here
4280 so that these comparisons behave in this particular context mostly
4281 like regular object comparisons. Thankfully Python is OK with
4282 that! Otherwise we'd have to use special set classes for columns
4283 (which we used to do, decades ago).
4285 """
4286 if self.operator in (operators.eq, operators.ne):
4287 # this is using the eq/ne operator given int hash values,
4288 # rather than Operator, so that "bool" can be based on
4289 # identity
4290 return self.operator(*self._orig) # type: ignore
4291 else:
4292 raise TypeError("Boolean value of this clause is not defined")
4294 if typing.TYPE_CHECKING:
4296 def __invert__(
4297 self: BinaryExpression[_T],
4298 ) -> BinaryExpression[_T]: ...
4300 @util.ro_non_memoized_property
4301 def _from_objects(self) -> List[FromClause]:
4302 return self.left._from_objects + self.right._from_objects
4304 def _negate(self):
4305 if self.negate is not None:
4306 return BinaryExpression(
4307 self.left,
4308 self.right._negate_in_binary(self.negate, self.operator),
4309 self.negate,
4310 negate=self.operator,
4311 type_=self.type,
4312 modifiers=self.modifiers,
4313 )
4314 else:
4315 return self.self_group()._negate()
4318class Slice(ColumnElement[Any]):
4319 """Represent SQL for a Python array-slice object.
4321 This is not a specific SQL construct at this level, but
4322 may be interpreted by specific dialects, e.g. PostgreSQL.
4324 """
4326 __visit_name__ = "slice"
4328 _traverse_internals: _TraverseInternalsType = [
4329 ("start", InternalTraversal.dp_clauseelement),
4330 ("stop", InternalTraversal.dp_clauseelement),
4331 ("step", InternalTraversal.dp_clauseelement),
4332 ]
4334 def __init__(self, start, stop, step, _name=None):
4335 self.start = coercions.expect(
4336 roles.ExpressionElementRole,
4337 start,
4338 name=_name,
4339 type_=type_api.INTEGERTYPE,
4340 )
4341 self.stop = coercions.expect(
4342 roles.ExpressionElementRole,
4343 stop,
4344 name=_name,
4345 type_=type_api.INTEGERTYPE,
4346 )
4347 self.step = coercions.expect(
4348 roles.ExpressionElementRole,
4349 step,
4350 name=_name,
4351 type_=type_api.INTEGERTYPE,
4352 )
4353 self.type = type_api.NULLTYPE
4355 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4356 assert against is operator.getitem
4357 return self
4360class IndexExpression(BinaryExpression[Any]):
4361 """Represent the class of expressions that are like an "index"
4362 operation."""
4364 inherit_cache = True
4367class GroupedElement(DQLDMLClauseElement):
4368 """Represent any parenthesized expression"""
4370 __visit_name__ = "grouping"
4372 def self_group(self, against: Optional[OperatorType] = None) -> Self:
4373 return self
4375 def _ungroup(self) -> ClauseElement:
4376 raise NotImplementedError()
4379class Grouping(GroupedElement, ColumnElement[_T]):
4380 """Represent a grouping within a column expression"""
4382 _traverse_internals: _TraverseInternalsType = [
4383 ("element", InternalTraversal.dp_clauseelement),
4384 ("type", InternalTraversal.dp_type),
4385 ]
4387 _cache_key_traversal = [
4388 ("element", InternalTraversal.dp_clauseelement),
4389 ]
4391 element: Union[
4392 AbstractTextClause,
4393 ClauseList,
4394 ColumnElement[_T],
4395 CompilerColumnElement,
4396 ]
4398 def __init__(
4399 self,
4400 element: Union[
4401 AbstractTextClause,
4402 ClauseList,
4403 ColumnElement[_T],
4404 CompilerColumnElement,
4405 ],
4406 ):
4407 self.element = element
4409 # nulltype assignment issue
4410 self.type = getattr(element, "type", type_api.NULLTYPE) # type: ignore
4411 self._propagate_attrs = element._propagate_attrs
4413 def _with_binary_element_type(self, type_):
4414 return self.__class__(self.element._with_binary_element_type(type_))
4416 def _ungroup(self) -> ColumnElement[_T]:
4417 assert isinstance(self.element, ColumnElement)
4418 return self.element._ungroup()
4420 @util.memoized_property
4421 def _is_implicitly_boolean(self):
4422 return self.element._is_implicitly_boolean
4424 @util.non_memoized_property
4425 def _tq_label(self) -> Optional[str]:
4426 return (
4427 getattr(self.element, "_tq_label", None) or self._anon_name_label
4428 )
4430 @util.non_memoized_property
4431 def _proxies(self) -> List[ColumnElement[Any]]:
4432 if isinstance(self.element, ColumnElement):
4433 return [self.element]
4434 else:
4435 return []
4437 @util.ro_non_memoized_property
4438 def _from_objects(self) -> List[FromClause]:
4439 return self.element._from_objects
4441 def __getattr__(self, attr):
4442 return getattr(self.element, attr)
4444 def __getstate__(self):
4445 return {"element": self.element, "type": self.type}
4447 def __setstate__(self, state):
4448 self.element = state["element"]
4449 self.type = state["type"]
4451 if TYPE_CHECKING:
4453 def self_group(
4454 self, against: Optional[OperatorType] = None
4455 ) -> Self: ...
4458class _OverrideBinds(Grouping[_T]):
4459 """used by cache_key->_apply_params_to_element to allow compilation /
4460 execution of a SQL element that's been cached, using an alternate set of
4461 bound parameter values.
4463 This is used by the ORM to swap new parameter values into expressions
4464 that are embedded into loader options like with_expression(),
4465 selectinload(). Previously, this task was accomplished using the
4466 .params() method which would perform a deep-copy instead. This deep
4467 copy proved to be too expensive for more complex expressions.
4469 See #11085
4471 """
4473 __visit_name__ = "override_binds"
4475 def __init__(
4476 self,
4477 element: ColumnElement[_T],
4478 bindparams: Sequence[BindParameter[Any]],
4479 replaces_params: Sequence[BindParameter[Any]],
4480 ):
4481 self.element = element
4482 self.translate = {
4483 k.key: v.value for k, v in zip(replaces_params, bindparams)
4484 }
4486 def _gen_cache_key(
4487 self, anon_map: anon_map, bindparams: List[BindParameter[Any]]
4488 ) -> Optional[typing_Tuple[Any, ...]]:
4489 """generate a cache key for the given element, substituting its bind
4490 values for the translation values present."""
4492 existing_bps: List[BindParameter[Any]] = []
4493 ck = self.element._gen_cache_key(anon_map, existing_bps)
4495 bindparams.extend(
4496 (
4497 bp._with_value(
4498 self.translate[bp.key], maintain_key=True, required=False
4499 )
4500 if bp.key in self.translate
4501 else bp
4502 )
4503 for bp in existing_bps
4504 )
4506 return ck
4509_FrameIntTuple = tuple[int | None, int | None]
4512class Over(ColumnElement[_T]):
4513 """Represent an OVER clause.
4515 This is a special operator against a so-called
4516 "window" function, as well as any aggregate function,
4517 which produces results relative to the result set
4518 itself. Most modern SQL backends now support window functions.
4520 """
4522 __visit_name__ = "over"
4524 _traverse_internals: _TraverseInternalsType = [
4525 ("element", InternalTraversal.dp_clauseelement),
4526 ("order_by", InternalTraversal.dp_clauseelement),
4527 ("partition_by", InternalTraversal.dp_clauseelement),
4528 ("range_", InternalTraversal.dp_clauseelement),
4529 ("rows", InternalTraversal.dp_clauseelement),
4530 ("groups", InternalTraversal.dp_clauseelement),
4531 ("exclude", InternalTraversal.dp_string),
4532 ]
4534 order_by: Optional[ClauseList] = None
4535 partition_by: Optional[ClauseList] = None
4537 element: ColumnElement[_T]
4538 """The underlying expression object to which this :class:`.Over`
4539 object refers."""
4541 range_: FrameClause | None
4542 rows: FrameClause | None
4543 groups: FrameClause | None
4544 exclude: str | None
4546 def __init__(
4547 self,
4548 element: ColumnElement[_T],
4549 partition_by: Optional[_ByArgument] = None,
4550 order_by: Optional[_ByArgument] = None,
4551 range_: _FrameIntTuple | FrameClause | None = None,
4552 rows: _FrameIntTuple | FrameClause | None = None,
4553 groups: _FrameIntTuple | FrameClause | None = None,
4554 exclude: str | None = None,
4555 ):
4556 self.element = element
4557 if order_by is not None:
4558 self.order_by = ClauseList(
4559 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4560 )
4561 if partition_by is not None:
4562 self.partition_by = ClauseList(
4563 *util.to_list(partition_by),
4564 _literal_as_text_role=roles.ByOfRole,
4565 )
4567 if sum(item is not None for item in (range_, rows, groups)) > 1:
4568 raise exc.ArgumentError(
4569 "only one of 'rows', 'range_', or 'groups' may be provided"
4570 )
4571 else:
4572 self.range_ = FrameClause._parse(range_, coerce_int=False)
4573 self.rows = FrameClause._parse(rows, coerce_int=True)
4574 self.groups = FrameClause._parse(groups, coerce_int=True)
4576 self.exclude = exclude
4578 if exclude is not None and (
4579 range_ is None and rows is None and groups is None
4580 ):
4581 raise exc.ArgumentError(
4582 "'exclude' requires that one of 'rows', "
4583 "'range_', or 'groups' is also specified"
4584 )
4586 if not TYPE_CHECKING:
4588 @util.memoized_property
4589 def type(self) -> TypeEngine[_T]: # noqa: A001
4590 return self.element.type
4592 @util.ro_non_memoized_property
4593 def _from_objects(self) -> List[FromClause]:
4594 return list(
4595 itertools.chain(
4596 *[
4597 c._from_objects
4598 for c in (self.element, self.partition_by, self.order_by)
4599 if c is not None
4600 ]
4601 )
4602 )
4605class FrameClauseType(Enum):
4606 """Frame clause type enum for FrameClause lower_type and upper_type.
4608 .. versionadded:: 2.1
4610 """
4612 UNBOUNDED = 0
4613 """Produces an "UNBOUNDED PRECEDING" or "UNBOUNDED FOLLOWING" frame
4614 clause depending on the position.
4615 Requires a ``None`` value for the corresponding bound value.
4616 """
4617 CURRENT = 1
4618 """Produces a "CURRENT ROW" frame clause.
4619 Requires a ``None`` value for the corresponding bound value.
4620 """
4621 PRECEDING = 2
4622 """Produces a "PRECEDING" frame clause."""
4623 FOLLOWING = 3
4624 """Produces a "FOLLOWING" frame clause."""
4627_require_none = (
4628 FrameClauseType.CURRENT,
4629 FrameClauseType.UNBOUNDED,
4630)
4633class FrameClause(ClauseElement):
4634 """Indicate the 'rows' 'range' or 'group' field of a window function,
4635 e.g. using :class:`.Over`.
4637 .. versionadded:: 2.1
4639 """
4641 __visit_name__ = "frame_clause"
4643 _traverse_internals: _TraverseInternalsType = [
4644 ("lower_bind", InternalTraversal.dp_clauseelement),
4645 ("upper_bind", InternalTraversal.dp_clauseelement),
4646 ("lower_type", InternalTraversal.dp_plain_obj),
4647 ("upper_type", InternalTraversal.dp_plain_obj),
4648 ]
4650 def __init__(
4651 self,
4652 start: Any,
4653 end: Any,
4654 start_frame_type: FrameClauseType,
4655 end_frame_type: FrameClauseType,
4656 _validate: bool = True,
4657 ) -> None:
4658 """Creates a new FrameClause specifying the bounds of a window frame.
4660 :param start: The start value.
4661 :param end: The end value.
4662 :param start_frame_type: The :class:`FrameClauseType` for the
4663 start value.
4664 :param end_frame_type: The :class:`FrameClauseType` for the end value.
4665 """
4666 self.lower_bind = self._as_literal(start)
4667 self.upper_bind = self._as_literal(end)
4668 self.lower_type = FrameClauseType(start_frame_type)
4669 self.upper_type = FrameClauseType(end_frame_type)
4670 if _validate:
4671 if (
4672 self.lower_type in _require_none
4673 and self.lower_bind is not None
4674 ):
4675 raise exc.ArgumentError(
4676 "Cannot specify a value for start with frame type "
4677 f"{self.lower_type.name}"
4678 )
4679 if (
4680 self.upper_type in _require_none
4681 and self.upper_bind is not None
4682 ):
4683 raise exc.ArgumentError(
4684 "Cannot specify a value for end with frame type "
4685 f"{self.upper_type.name}"
4686 )
4688 @classmethod
4689 def _as_literal(cls, value: Any) -> BindParameter[Any] | None:
4690 if value is None:
4691 return None
4692 elif isinstance(value, int):
4693 return literal(value, type_api.INTEGERTYPE)
4694 elif isinstance(value, BindParameter):
4695 return value
4696 else:
4697 return literal(value) # let the default type resolution occur
4699 @classmethod
4700 def _handle_int(
4701 cls, value: Any | None, coerce_int: bool
4702 ) -> tuple[int | None, FrameClauseType]:
4703 if value is None:
4704 return None, FrameClauseType.UNBOUNDED
4706 if coerce_int:
4707 try:
4708 integer = int(value)
4709 except ValueError as err:
4710 raise exc.ArgumentError(
4711 "Integer or None expected for values in rows/groups frame"
4712 ) from err
4713 elif not isinstance(value, int):
4714 raise exc.ArgumentError(
4715 "When using a tuple to specify a range only integer or none "
4716 "values are allowed in the range frame. To specify a "
4717 "different type use the FrameClause directly."
4718 )
4719 else:
4720 integer = value
4721 if integer == 0:
4722 return None, FrameClauseType.CURRENT
4723 elif integer < 0:
4724 return abs(integer), FrameClauseType.PRECEDING
4725 else:
4726 return integer, FrameClauseType.FOLLOWING
4728 @classmethod
4729 def _parse(
4730 cls,
4731 range_: _FrameIntTuple | FrameClause | None,
4732 coerce_int: bool,
4733 ) -> FrameClause | None:
4734 if range_ is None or isinstance(range_, FrameClause):
4735 return range_
4737 try:
4738 r0, r1 = range_
4739 except (ValueError, TypeError) as ve:
4740 raise exc.ArgumentError(
4741 "2-tuple expected for range/rows/groups"
4742 ) from ve
4744 l_b, l_t = cls._handle_int(r0, coerce_int)
4745 u_b, u_t = cls._handle_int(r1, coerce_int)
4747 return FrameClause(
4748 start=l_b,
4749 end=u_b,
4750 start_frame_type=l_t,
4751 end_frame_type=u_t,
4752 _validate=False,
4753 )
4756class AggregateOrderBy(WrapsColumnExpression[_T]):
4757 """Represent an aggregate ORDER BY expression.
4759 This is a special operator against aggregate functions such as
4760 ``array_agg()``, ``json_arrayagg()`` ``string_agg()``, etc. that provides
4761 for an ORDER BY expression, using a syntax that's compatible with
4762 the backend.
4764 :class:`.AggregateOrderBy` is a generalized version of the
4765 :class:`.WithinGroup` construct, the latter of which always provides a
4766 "WITHIN GROUP (ORDER BY ...)" expression. :class:`.AggregateOrderBy` will
4767 also compile to "WITHIN GROUP (ORDER BY ...)" on backends such as Oracle
4768 and SQL Server that don't have another style of aggregate function
4769 ordering.
4771 .. versionadded:: 2.1
4774 """
4776 __visit_name__ = "aggregateorderby"
4778 _traverse_internals: _TraverseInternalsType = [
4779 ("element", InternalTraversal.dp_clauseelement),
4780 ("order_by", InternalTraversal.dp_clauseelement),
4781 ]
4783 order_by: ClauseList
4785 def __init__(
4786 self,
4787 element: Union[FunctionElement[_T], FunctionFilter[_T]],
4788 *order_by: _ColumnExpressionArgument[Any],
4789 ):
4790 self.element = element
4791 if not order_by:
4792 raise TypeError("at least one ORDER BY element is required")
4793 self.order_by = ClauseList(
4794 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4795 )
4797 if not TYPE_CHECKING:
4799 @util.memoized_property
4800 def type(self) -> TypeEngine[_T]: # noqa: A001
4801 return self.element.type
4803 @property
4804 def wrapped_column_expression(self) -> ColumnElement[_T]:
4805 return self.element
4807 def __reduce__(self):
4808 return self.__class__, (self.element,) + (
4809 tuple(self.order_by) if self.order_by is not None else ()
4810 )
4812 def over(
4813 self,
4814 *,
4815 partition_by: _ByArgument | None = None,
4816 order_by: _ByArgument | None = None,
4817 rows: _FrameIntTuple | FrameClause | None = None,
4818 range_: _FrameIntTuple | FrameClause | None = None,
4819 groups: _FrameIntTuple | FrameClause | None = None,
4820 exclude: str | None = None,
4821 ) -> Over[_T]:
4822 """Produce an OVER clause against this :class:`.WithinGroup`
4823 construct.
4825 This function has the same signature as that of
4826 :meth:`.FunctionElement.over`.
4828 """
4829 return Over(
4830 self,
4831 partition_by=partition_by,
4832 order_by=order_by,
4833 range_=range_,
4834 rows=rows,
4835 groups=groups,
4836 exclude=exclude,
4837 )
4839 @overload
4840 def filter(self) -> Self: ...
4842 @overload
4843 def filter(
4844 self,
4845 __criterion0: _ColumnExpressionArgument[bool],
4846 *criterion: _ColumnExpressionArgument[bool],
4847 ) -> FunctionFilter[_T]: ...
4849 def filter(
4850 self, *criterion: _ColumnExpressionArgument[bool]
4851 ) -> Union[Self, FunctionFilter[_T]]:
4852 """Produce a FILTER clause against this function."""
4853 if not criterion:
4854 return self
4855 return FunctionFilter(self, *criterion)
4857 @util.ro_non_memoized_property
4858 def _from_objects(self) -> List[FromClause]:
4859 return list(
4860 itertools.chain(
4861 *[
4862 c._from_objects
4863 for c in (self.element, self.order_by)
4864 if c is not None
4865 ]
4866 )
4867 )
4870class WithinGroup(AggregateOrderBy[_T]):
4871 """Represent a WITHIN GROUP (ORDER BY) clause.
4873 This is a special operator against so-called
4874 "ordered set aggregate" and "hypothetical
4875 set aggregate" functions, including ``percentile_cont()``,
4876 ``rank()``, ``dense_rank()``, etc.
4878 It's supported only by certain database backends, such as PostgreSQL,
4879 Oracle Database and MS SQL Server.
4881 The :class:`.WithinGroup` construct extracts its type from the
4882 method :meth:`.FunctionElement.within_group_type`. If this returns
4883 ``None``, the function's ``.type`` is used.
4885 """
4887 __visit_name__ = "withingroup"
4888 inherit_cache = True
4890 if not TYPE_CHECKING:
4892 @util.memoized_property
4893 def type(self) -> TypeEngine[_T]: # noqa: A001
4894 wgt = self.element.within_group_type(self)
4895 if wgt is not None:
4896 return wgt
4897 else:
4898 return self.element.type
4901class FunctionFilter(Generative, ColumnElement[_T]):
4902 """Represent a function FILTER clause.
4904 This is a special operator against aggregate and window functions,
4905 which controls which rows are passed to it.
4906 It's supported only by certain database backends.
4908 Invocation of :class:`.FunctionFilter` is via
4909 :meth:`.FunctionElement.filter`::
4911 func.count(1).filter(True)
4913 .. seealso::
4915 :meth:`.FunctionElement.filter`
4917 """
4919 __visit_name__ = "funcfilter"
4921 _traverse_internals: _TraverseInternalsType = [
4922 ("func", InternalTraversal.dp_clauseelement),
4923 ("criterion", InternalTraversal.dp_clauseelement),
4924 ]
4926 criterion: Optional[ColumnElement[bool]] = None
4928 def __init__(
4929 self,
4930 func: Union[FunctionElement[_T], AggregateOrderBy[_T]],
4931 *criterion: _ColumnExpressionArgument[bool],
4932 ):
4933 self.func = func
4934 self.filter.non_generative(self, *criterion) # type: ignore
4936 @_generative
4937 def filter(self, *criterion: _ColumnExpressionArgument[bool]) -> Self:
4938 """Produce an additional FILTER against the function.
4940 This method adds additional criteria to the initial criteria
4941 set up by :meth:`.FunctionElement.filter`.
4943 Multiple criteria are joined together at SQL render time
4944 via ``AND``.
4947 """
4949 for crit in list(criterion):
4950 crit = coercions.expect(roles.WhereHavingRole, crit)
4952 if self.criterion is not None:
4953 self.criterion = self.criterion & crit
4954 else:
4955 self.criterion = crit
4957 return self
4959 def over(
4960 self,
4961 partition_by: _ByArgument | None = None,
4962 order_by: _ByArgument | None = None,
4963 range_: _FrameIntTuple | FrameClause | None = None,
4964 rows: _FrameIntTuple | FrameClause | None = None,
4965 groups: _FrameIntTuple | FrameClause | None = None,
4966 exclude: str | None = None,
4967 ) -> Over[_T]:
4968 """Produce an OVER clause against this filtered function.
4970 Used against aggregate or so-called "window" functions,
4971 for database backends that support window functions.
4973 The expression::
4975 func.rank().filter(MyClass.y > 5).over(order_by="x")
4977 is shorthand for::
4979 from sqlalchemy import over, funcfilter
4981 over(funcfilter(func.rank(), MyClass.y > 5), order_by="x")
4983 See :func:`_expression.over` for a full description.
4985 """
4986 return Over(
4987 self,
4988 partition_by=partition_by,
4989 order_by=order_by,
4990 range_=range_,
4991 rows=rows,
4992 groups=groups,
4993 exclude=exclude,
4994 )
4996 def within_group(
4997 self, *order_by: _ColumnExpressionArgument[Any]
4998 ) -> WithinGroup[_T]:
4999 """Produce a WITHIN GROUP (ORDER BY expr) clause against
5000 this function.
5001 """
5002 return WithinGroup(self, *order_by)
5004 def within_group_type(
5005 self, within_group: WithinGroup[_T]
5006 ) -> Optional[TypeEngine[_T]]:
5007 return None
5009 def self_group(
5010 self, against: Optional[OperatorType] = None
5011 ) -> Union[Self, Grouping[_T]]:
5012 if operators.is_precedent(operators.filter_op, against):
5013 return Grouping(self)
5014 else:
5015 return self
5017 if not TYPE_CHECKING:
5019 @util.memoized_property
5020 def type(self) -> TypeEngine[_T]: # noqa: A001
5021 return self.func.type
5023 @util.ro_non_memoized_property
5024 def _from_objects(self) -> List[FromClause]:
5025 return list(
5026 itertools.chain(
5027 *[
5028 c._from_objects
5029 for c in (self.func, self.criterion)
5030 if c is not None
5031 ]
5032 )
5033 )
5036class NamedColumn(KeyedColumnElement[_T]):
5037 is_literal = False
5038 table: Optional[FromClause] = None
5039 name: str
5040 key: str
5042 def _compare_name_for_result(self, other):
5043 return (hasattr(other, "name") and self.name == other.name) or (
5044 hasattr(other, "_label") and self._label == other._label
5045 )
5047 @util.ro_memoized_property
5048 def description(self) -> str:
5049 return self.name
5051 @HasMemoized.memoized_attribute
5052 def _tq_key_label(self) -> Optional[str]:
5053 """table qualified label based on column key.
5055 for table-bound columns this is <tablename>_<column key/proxy key>;
5057 all other expressions it resolves to key/proxy key.
5059 """
5060 proxy_key = self._proxy_key
5061 if proxy_key and proxy_key != self.name:
5062 return self._gen_tq_label(proxy_key)
5063 else:
5064 return self._tq_label
5066 @HasMemoized.memoized_attribute
5067 def _tq_label(self) -> Optional[str]:
5068 """table qualified label based on column name.
5070 for table-bound columns this is <tablename>_<columnname>; all other
5071 expressions it resolves to .name.
5073 """
5074 return self._gen_tq_label(self.name)
5076 @HasMemoized.memoized_attribute
5077 def _render_label_in_columns_clause(self):
5078 return True
5080 @HasMemoized.memoized_attribute
5081 def _non_anon_label(self):
5082 return self.name
5084 def _gen_tq_label(
5085 self, name: str, dedupe_on_key: bool = True
5086 ) -> Optional[str]:
5087 return name
5089 def _bind_param(
5090 self,
5091 operator: OperatorType,
5092 obj: Any,
5093 type_: Optional[TypeEngine[_T]] = None,
5094 expanding: bool = False,
5095 ) -> BindParameter[_T]:
5096 return BindParameter(
5097 self.key,
5098 obj,
5099 _compared_to_operator=operator,
5100 _compared_to_type=self.type,
5101 type_=type_,
5102 unique=True,
5103 expanding=expanding,
5104 )
5106 def _make_proxy(
5107 self,
5108 selectable: FromClause,
5109 *,
5110 primary_key: ColumnSet,
5111 foreign_keys: Set[KeyedColumnElement[Any]],
5112 name: Optional[str] = None,
5113 key: Optional[str] = None,
5114 name_is_truncatable: bool = False,
5115 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5116 disallow_is_literal: bool = False,
5117 **kw: Any,
5118 ) -> typing_Tuple[str, ColumnClause[_T]]:
5119 c = ColumnClause(
5120 (
5121 coercions.expect(roles.TruncatedLabelRole, name or self.name)
5122 if name_is_truncatable
5123 else (name or self.name)
5124 ),
5125 type_=self.type,
5126 _selectable=selectable,
5127 is_literal=False,
5128 )
5130 c._propagate_attrs = selectable._propagate_attrs
5131 if name is None:
5132 c.key = self.key
5133 if compound_select_cols:
5134 c._proxies = list(compound_select_cols)
5135 else:
5136 c._proxies = [self]
5138 if selectable._is_clone_of is not None:
5139 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5140 return c.key, c
5143_PS = ParamSpec("_PS")
5146class Label(roles.LabeledColumnExprRole[_T], NamedColumn[_T]):
5147 """Represents a column label (AS).
5149 Represent a label, as typically applied to any column-level
5150 element using the ``AS`` sql keyword.
5152 """
5154 __visit_name__ = "label"
5156 _traverse_internals: _TraverseInternalsType = [
5157 ("name", InternalTraversal.dp_anon_name),
5158 ("type", InternalTraversal.dp_type),
5159 ("_element", InternalTraversal.dp_clauseelement),
5160 ]
5162 _cache_key_traversal = [
5163 ("name", InternalTraversal.dp_anon_name),
5164 ("_element", InternalTraversal.dp_clauseelement),
5165 ]
5167 _element: ColumnElement[_T]
5168 name: str
5170 def __init__(
5171 self,
5172 name: Optional[str],
5173 element: _ColumnExpressionArgument[_T],
5174 type_: Optional[_TypeEngineArgument[_T]] = None,
5175 ):
5176 orig_element = element
5177 element = coercions.expect(
5178 roles.ExpressionElementRole,
5179 element,
5180 apply_propagate_attrs=self,
5181 )
5182 while isinstance(element, Label):
5183 # TODO: this is only covered in test_text.py, but nothing
5184 # fails if it's removed. determine rationale
5185 element = element.element
5187 if name:
5188 self.name = name
5189 else:
5190 self.name = _anonymous_label.safe_construct(
5191 id(self), getattr(element, "name", "anon")
5192 )
5193 if isinstance(orig_element, Label):
5194 # TODO: no coverage for this block, again would be in
5195 # test_text.py where the resolve_label concept is important
5196 self._resolve_label = orig_element._label
5198 self.key = self._tq_label = self._tq_key_label = self.name
5199 self._element = element
5201 self.type = (
5202 type_api.to_instance(type_)
5203 if type_ is not None
5204 else self._element.type
5205 )
5207 self._proxies = [element]
5209 def __reduce__(self):
5210 return self.__class__, (self.name, self._element, self.type)
5212 @HasMemoized.memoized_attribute
5213 def _render_label_in_columns_clause(self):
5214 return True
5216 def _bind_param(self, operator, obj, type_=None, expanding=False):
5217 return BindParameter(
5218 None,
5219 obj,
5220 _compared_to_operator=operator,
5221 type_=type_,
5222 _compared_to_type=self.type,
5223 unique=True,
5224 expanding=expanding,
5225 )
5227 @util.memoized_property
5228 def _is_implicitly_boolean(self):
5229 return self.element._is_implicitly_boolean
5231 @HasMemoized.memoized_attribute
5232 def _allow_label_resolve(self):
5233 return self.element._allow_label_resolve
5235 @property
5236 def _order_by_label_element(self):
5237 return self
5239 def as_reference(self) -> _label_reference[_T]:
5240 """refer to this labeled expression in a clause such as GROUP BY,
5241 ORDER BY etc. as the label name itself, without expanding
5242 into the full expression.
5244 .. versionadded:: 2.1
5246 """
5247 return _label_reference(self)
5249 @HasMemoized.memoized_attribute
5250 def element(self) -> ColumnElement[_T]:
5251 return self._element.self_group(against=operators.as_)
5253 def self_group(self, against: Optional[OperatorType] = None) -> Label[_T]:
5254 return self._apply_to_inner(self._element.self_group, against=against)
5256 def _negate(self):
5257 return self._apply_to_inner(self._element._negate)
5259 def _apply_to_inner(
5260 self,
5261 fn: Callable[_PS, ColumnElement[_T]],
5262 *arg: _PS.args,
5263 **kw: _PS.kwargs,
5264 ) -> Label[_T]:
5265 sub_element = fn(*arg, **kw)
5266 if sub_element is not self._element:
5267 return Label(self.name, sub_element, type_=self.type)
5268 else:
5269 return self
5271 @property
5272 def primary_key(self): # type: ignore[override]
5273 return self.element.primary_key
5275 @property
5276 def foreign_keys(self): # type: ignore[override]
5277 return self.element.foreign_keys
5279 def _copy_internals(
5280 self,
5281 *,
5282 clone: _CloneCallableType = _clone,
5283 anonymize_labels: bool = False,
5284 **kw: Any,
5285 ) -> None:
5286 self._reset_memoizations()
5287 self._element = clone(self._element, **kw)
5288 if anonymize_labels:
5289 self.name = _anonymous_label.safe_construct(
5290 id(self), getattr(self.element, "name", "anon")
5291 )
5292 self.key = self._tq_label = self._tq_key_label = self.name
5294 @util.ro_non_memoized_property
5295 def _from_objects(self) -> List[FromClause]:
5296 return self.element._from_objects
5298 def _make_proxy(
5299 self,
5300 selectable: FromClause,
5301 *,
5302 primary_key: ColumnSet,
5303 foreign_keys: Set[KeyedColumnElement[Any]],
5304 name: Optional[str] = None,
5305 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5306 **kw: Any,
5307 ) -> typing_Tuple[str, ColumnClause[_T]]:
5308 name = self.name if not name else name
5310 key, e = self.element._make_proxy(
5311 selectable,
5312 name=name,
5313 disallow_is_literal=True,
5314 name_is_truncatable=isinstance(name, _truncated_label),
5315 compound_select_cols=compound_select_cols,
5316 primary_key=primary_key,
5317 foreign_keys=foreign_keys,
5318 )
5320 # there was a note here to remove this assertion, which was here
5321 # to determine if we later could support a use case where
5322 # the key and name of a label are separate. But I don't know what
5323 # that case was. For now, this is an unexpected case that occurs
5324 # when a label name conflicts with other columns and select()
5325 # is attempting to disambiguate an explicit label, which is not what
5326 # the user would want. See issue #6090.
5327 if key != self.name and not isinstance(self.name, _anonymous_label):
5328 raise exc.InvalidRequestError(
5329 "Label name %s is being renamed to an anonymous label due "
5330 "to disambiguation "
5331 "which is not supported right now. Please use unique names "
5332 "for explicit labels." % (self.name)
5333 )
5335 e._propagate_attrs = selectable._propagate_attrs
5336 e._proxies.append(self)
5337 if self.type is not None:
5338 e.type = self.type
5340 return self.key, e
5343class ColumnClause(
5344 roles.DDLReferredColumnRole,
5345 roles.LabeledColumnExprRole[_T],
5346 roles.StrAsPlainColumnRole,
5347 Immutable,
5348 NamedColumn[_T],
5349):
5350 """Represents a column expression from any textual string.
5352 The :class:`.ColumnClause`, a lightweight analogue to the
5353 :class:`_schema.Column` class, is typically invoked using the
5354 :func:`_expression.column` function, as in::
5356 from sqlalchemy import column
5358 id, name = column("id"), column("name")
5359 stmt = select(id, name).select_from("user")
5361 The above statement would produce SQL like:
5363 .. sourcecode:: sql
5365 SELECT id, name FROM user
5367 :class:`.ColumnClause` is the immediate superclass of the schema-specific
5368 :class:`_schema.Column` object. While the :class:`_schema.Column`
5369 class has all the
5370 same capabilities as :class:`.ColumnClause`, the :class:`.ColumnClause`
5371 class is usable by itself in those cases where behavioral requirements
5372 are limited to simple SQL expression generation. The object has none of
5373 the associations with schema-level metadata or with execution-time
5374 behavior that :class:`_schema.Column` does,
5375 so in that sense is a "lightweight"
5376 version of :class:`_schema.Column`.
5378 Full details on :class:`.ColumnClause` usage is at
5379 :func:`_expression.column`.
5381 .. seealso::
5383 :func:`_expression.column`
5385 :class:`_schema.Column`
5387 """
5389 table: Optional[FromClause]
5390 is_literal: bool
5392 __visit_name__ = "column"
5394 _traverse_internals: _TraverseInternalsType = [
5395 ("name", InternalTraversal.dp_anon_name),
5396 ("type", InternalTraversal.dp_type),
5397 ("table", InternalTraversal.dp_clauseelement),
5398 ("is_literal", InternalTraversal.dp_boolean),
5399 ]
5401 onupdate: Optional[DefaultGenerator] = None
5402 default: Optional[DefaultGenerator] = None
5403 server_default: Optional[FetchedValue] = None
5404 server_onupdate: Optional[FetchedValue] = None
5406 _is_multiparam_column = False
5408 @property
5409 def _is_star(self): # type: ignore[override]
5410 return self.is_literal and self.name == "*"
5412 def __init__(
5413 self,
5414 text: str,
5415 type_: Optional[_TypeEngineArgument[_T]] = None,
5416 is_literal: bool = False,
5417 _selectable: Optional[FromClause] = None,
5418 ):
5419 self.key = self.name = text
5420 self.table = _selectable
5422 # if type is None, we get NULLTYPE, which is our _T. But I don't
5423 # know how to get the overloads to express that correctly
5424 self.type = type_api.to_instance(type_) # type: ignore
5426 self.is_literal = is_literal
5428 def get_children(self, *, column_tables=False, **kw):
5429 # override base get_children() to not return the Table
5430 # or selectable that is parent to this column. Traversals
5431 # expect the columns of tables and subqueries to be leaf nodes.
5432 return []
5434 @property
5435 def entity_namespace(self):
5436 if self.table is not None:
5437 return self.table.entity_namespace
5438 else:
5439 return super().entity_namespace
5441 def _clone(self, detect_subquery_cols=False, **kw):
5442 if (
5443 detect_subquery_cols
5444 and self.table is not None
5445 and self.table._is_subquery
5446 ):
5447 clone = kw.pop("clone")
5448 table = clone(self.table, **kw)
5449 new = table.c.corresponding_column(self)
5450 return new
5452 return super()._clone(**kw)
5454 @HasMemoized_ro_memoized_attribute
5455 def _from_objects(self) -> List[FromClause]:
5456 t = self.table
5457 if t is not None:
5458 return [t]
5459 else:
5460 return []
5462 @HasMemoized.memoized_attribute
5463 def _render_label_in_columns_clause(self):
5464 return self.table is not None
5466 @property
5467 def _ddl_label(self):
5468 return self._gen_tq_label(self.name, dedupe_on_key=False)
5470 def _compare_name_for_result(self, other):
5471 if (
5472 self.is_literal
5473 or self.table is None
5474 or self.table._is_textual
5475 or not hasattr(other, "proxy_set")
5476 or (
5477 isinstance(other, ColumnClause)
5478 and (
5479 other.is_literal
5480 or other.table is None
5481 or other.table._is_textual
5482 )
5483 )
5484 ):
5485 return (hasattr(other, "name") and self.name == other.name) or (
5486 hasattr(other, "_tq_label")
5487 and self._tq_label == other._tq_label
5488 )
5489 else:
5490 return other.proxy_set.intersection(self.proxy_set)
5492 def _gen_tq_label(
5493 self, name: str, dedupe_on_key: bool = True
5494 ) -> Optional[str]:
5495 """generate table-qualified label
5497 for a table-bound column this is <tablename>_<columnname>.
5499 used primarily for LABEL_STYLE_TABLENAME_PLUS_COL
5500 as well as the .columns collection on a Join object.
5502 """
5503 label: str
5504 t = self.table
5505 if self.is_literal:
5506 return None
5507 elif t is not None and is_named_from_clause(t):
5508 if has_schema_attr(t) and t.schema:
5509 label = (
5510 t.schema.replace(".", "_") + "_" + t.name + ("_" + name)
5511 )
5512 else:
5513 assert not TYPE_CHECKING or isinstance(t, NamedFromClause)
5514 label = t.name + ("_" + name)
5516 # propagate name quoting rules for labels.
5517 if is_quoted_name(name) and name.quote is not None:
5518 if is_quoted_name(label):
5519 label.quote = name.quote
5520 else:
5521 label = quoted_name(label, name.quote)
5522 elif is_quoted_name(t.name) and t.name.quote is not None:
5523 # can't get this situation to occur, so let's
5524 # assert false on it for now
5525 assert not isinstance(label, quoted_name)
5526 label = quoted_name(label, t.name.quote)
5528 if dedupe_on_key:
5529 # ensure the label name doesn't conflict with that of an
5530 # existing column. note that this implies that any Column
5531 # must **not** set up its _label before its parent table has
5532 # all of its other Column objects set up. There are several
5533 # tables in the test suite which will fail otherwise; example:
5534 # table "owner" has columns "name" and "owner_name". Therefore
5535 # column owner.name cannot use the label "owner_name", it has
5536 # to be "owner_name_1".
5537 if label in t.c:
5538 _label = label
5539 counter = 1
5540 while _label in t.c:
5541 _label = label + f"_{counter}"
5542 counter += 1
5543 label = _label
5545 return coercions.expect(roles.TruncatedLabelRole, label)
5547 else:
5548 return name
5550 def _make_proxy(
5551 self,
5552 selectable: FromClause,
5553 *,
5554 primary_key: ColumnSet,
5555 foreign_keys: Set[KeyedColumnElement[Any]],
5556 name: Optional[str] = None,
5557 key: Optional[str] = None,
5558 name_is_truncatable: bool = False,
5559 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5560 disallow_is_literal: bool = False,
5561 **kw: Any,
5562 ) -> typing_Tuple[str, ColumnClause[_T]]:
5563 # the "is_literal" flag normally should never be propagated; a proxied
5564 # column is always a SQL identifier and never the actual expression
5565 # being evaluated. however, there is a case where the "is_literal" flag
5566 # might be used to allow the given identifier to have a fixed quoting
5567 # pattern already, so maintain the flag for the proxy unless a
5568 # :class:`.Label` object is creating the proxy. See [ticket:4730].
5569 is_literal = (
5570 not disallow_is_literal
5571 and self.is_literal
5572 and (
5573 # note this does not accommodate for quoted_name differences
5574 # right now
5575 name is None
5576 or name == self.name
5577 )
5578 )
5579 c = self._constructor(
5580 (
5581 coercions.expect(roles.TruncatedLabelRole, name or self.name)
5582 if name_is_truncatable
5583 else (name or self.name)
5584 ),
5585 type_=self.type,
5586 _selectable=selectable,
5587 is_literal=is_literal,
5588 )
5589 c._propagate_attrs = selectable._propagate_attrs
5590 if name is None:
5591 c.key = self.key
5592 if compound_select_cols:
5593 c._proxies = list(compound_select_cols)
5594 else:
5595 c._proxies = [self]
5597 if selectable._is_clone_of is not None:
5598 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5599 return c.key, c
5602class TableValuedColumn(NamedColumn[_T]):
5603 __visit_name__ = "table_valued_column"
5605 _traverse_internals: _TraverseInternalsType = [
5606 ("name", InternalTraversal.dp_anon_name),
5607 ("type", InternalTraversal.dp_type),
5608 ("scalar_alias", InternalTraversal.dp_clauseelement),
5609 ]
5611 def __init__(self, scalar_alias: NamedFromClause, type_: TypeEngine[_T]):
5612 self.scalar_alias = scalar_alias
5613 self.key = self.name = scalar_alias.name
5614 self.type = type_
5616 def _copy_internals(
5617 self, clone: _CloneCallableType = _clone, **kw: Any
5618 ) -> None:
5619 self.scalar_alias = clone(self.scalar_alias, **kw)
5620 self.key = self.name = self.scalar_alias.name
5622 @util.ro_non_memoized_property
5623 def _from_objects(self) -> List[FromClause]:
5624 return [self.scalar_alias]
5627class CollationClause(ColumnElement[str]):
5628 __visit_name__ = "collation"
5630 _traverse_internals: _TraverseInternalsType = [
5631 ("collation", InternalTraversal.dp_string)
5632 ]
5634 @classmethod
5635 @util.preload_module("sqlalchemy.sql.sqltypes")
5636 def _create_collation_expression(
5637 cls, expression: _ColumnExpressionArgument[str], collation: str
5638 ) -> BinaryExpression[str]:
5640 sqltypes = util.preloaded.sql_sqltypes
5642 expr = coercions.expect(roles.ExpressionElementRole[str], expression)
5644 if expr.type._type_affinity is sqltypes.String:
5645 collate_type = expr.type._with_collation(collation)
5646 else:
5647 collate_type = expr.type
5649 return BinaryExpression(
5650 expr,
5651 CollationClause(collation),
5652 operators.collate,
5653 type_=collate_type,
5654 )
5656 def __init__(self, collation):
5657 self.collation = collation
5660class _IdentifiedClause(Executable, ClauseElement):
5661 __visit_name__ = "identified"
5663 def __init__(self, ident):
5664 self.ident = ident
5667class SavepointClause(_IdentifiedClause):
5668 __visit_name__ = "savepoint"
5669 inherit_cache = False
5672class RollbackToSavepointClause(_IdentifiedClause):
5673 __visit_name__ = "rollback_to_savepoint"
5674 inherit_cache = False
5677class ReleaseSavepointClause(_IdentifiedClause):
5678 __visit_name__ = "release_savepoint"
5679 inherit_cache = False
5682class quoted_name(util.MemoizedSlots, str):
5683 """Represent a SQL identifier combined with quoting preferences.
5685 :class:`.quoted_name` is a Python unicode/str subclass which
5686 represents a particular identifier name along with a
5687 ``quote`` flag. This ``quote`` flag, when set to
5688 ``True`` or ``False``, overrides automatic quoting behavior
5689 for this identifier in order to either unconditionally quote
5690 or to not quote the name. If left at its default of ``None``,
5691 quoting behavior is applied to the identifier on a per-backend basis
5692 based on an examination of the token itself.
5694 A :class:`.quoted_name` object with ``quote=True`` is also
5695 prevented from being modified in the case of a so-called
5696 "name normalize" option. Certain database backends, such as
5697 Oracle Database, Firebird, and DB2 "normalize" case-insensitive names
5698 as uppercase. The SQLAlchemy dialects for these backends
5699 convert from SQLAlchemy's lower-case-means-insensitive convention
5700 to the upper-case-means-insensitive conventions of those backends.
5701 The ``quote=True`` flag here will prevent this conversion from occurring
5702 to support an identifier that's quoted as all lower case against
5703 such a backend.
5705 The :class:`.quoted_name` object is normally created automatically
5706 when specifying the name for key schema constructs such as
5707 :class:`_schema.Table`, :class:`_schema.Column`, and others.
5708 The class can also be
5709 passed explicitly as the name to any function that receives a name which
5710 can be quoted. Such as to use the :meth:`_engine.Engine.has_table`
5711 method with
5712 an unconditionally quoted name::
5714 from sqlalchemy import create_engine
5715 from sqlalchemy import inspect
5716 from sqlalchemy.sql import quoted_name
5718 engine = create_engine("oracle+oracledb://some_dsn")
5719 print(inspect(engine).has_table(quoted_name("some_table", True)))
5721 The above logic will run the "has table" logic against the Oracle Database
5722 backend, passing the name exactly as ``"some_table"`` without converting to
5723 upper case.
5725 """
5727 __slots__ = "quote", "lower", "upper"
5729 quote: Optional[bool]
5731 @overload
5732 @classmethod
5733 def construct(cls, value: str, quote: Optional[bool]) -> quoted_name: ...
5735 @overload
5736 @classmethod
5737 def construct(cls, value: None, quote: Optional[bool]) -> None: ...
5739 @classmethod
5740 def construct(
5741 cls, value: Optional[str], quote: Optional[bool]
5742 ) -> Optional[quoted_name]:
5743 if value is None:
5744 return None
5745 else:
5746 return quoted_name(value, quote)
5748 def __new__(cls, value: str, quote: Optional[bool]) -> quoted_name:
5749 assert (
5750 value is not None
5751 ), "use quoted_name.construct() for None passthrough"
5752 if isinstance(value, cls) and (quote is None or value.quote == quote):
5753 return value
5754 self = super().__new__(cls, value)
5756 self.quote = quote
5757 return self
5759 def __reduce__(self):
5760 return quoted_name, (str(self), self.quote)
5762 def _memoized_method_lower(self):
5763 if self.quote:
5764 return self
5765 else:
5766 return str(self).lower()
5768 def _memoized_method_upper(self):
5769 if self.quote:
5770 return self
5771 else:
5772 return str(self).upper()
5775def _find_columns(clause: ClauseElement) -> Set[ColumnClause[Any]]:
5776 """locate Column objects within the given expression."""
5778 cols: Set[ColumnClause[Any]] = set()
5779 traverse(clause, {}, {"column": cols.add})
5780 return cols
5783def _type_from_args(args: Sequence[ColumnElement[_T]]) -> TypeEngine[_T]:
5784 for a in args:
5785 if not a.type._isnull:
5786 return a.type
5787 else:
5788 return type_api.NULLTYPE # type: ignore
5791def _corresponding_column_or_error(fromclause, column, require_embedded=False):
5792 c = fromclause.corresponding_column(
5793 column, require_embedded=require_embedded
5794 )
5795 if c is None:
5796 raise exc.InvalidRequestError(
5797 "Given column '%s', attached to table '%s', "
5798 "failed to locate a corresponding column from table '%s'"
5799 % (column, getattr(column, "table", None), fromclause.description)
5800 )
5801 return c
5804class _memoized_property_but_not_nulltype(
5805 util.memoized_property["TypeEngine[_T]"]
5806):
5807 """memoized property, but dont memoize NullType"""
5809 def __get__(self, obj, cls):
5810 if obj is None:
5811 return self
5812 result = self.fget(obj)
5813 if not result._isnull:
5814 obj.__dict__[self.__name__] = result
5815 return result
5818class AnnotatedColumnElement(Annotated):
5819 _Annotated__element: ColumnElement[Any]
5821 def __init__(self, element, values):
5822 Annotated.__init__(self, element, values)
5823 for attr in (
5824 "comparator",
5825 "_proxy_key",
5826 "_tq_key_label",
5827 "_tq_label",
5828 "_non_anon_label",
5829 "type",
5830 ):
5831 self.__dict__.pop(attr, None)
5832 for attr in ("name", "key", "table"):
5833 if self.__dict__.get(attr, False) is None:
5834 self.__dict__.pop(attr)
5836 def _with_annotations(self, values):
5837 clone = super()._with_annotations(values)
5838 for attr in (
5839 "comparator",
5840 "_proxy_key",
5841 "_tq_key_label",
5842 "_tq_label",
5843 "_non_anon_label",
5844 ):
5845 clone.__dict__.pop(attr, None)
5846 return clone
5848 @util.memoized_property
5849 def name(self):
5850 """pull 'name' from parent, if not present"""
5851 return self._Annotated__element.name
5853 @_memoized_property_but_not_nulltype
5854 def type(self):
5855 """pull 'type' from parent and don't cache if null.
5857 type is routinely changed on existing columns within the
5858 mapped_column() initialization process, and "type" is also consulted
5859 during the creation of SQL expressions. Therefore it can change after
5860 it was already retrieved. At the same time we don't want annotated
5861 objects having overhead when expressions are produced, so continue
5862 to memoize, but only when we have a non-null type.
5864 """
5865 return self._Annotated__element.type
5867 @util.memoized_property
5868 def table(self):
5869 """pull 'table' from parent, if not present"""
5870 return self._Annotated__element.table
5872 @util.memoized_property
5873 def key(self):
5874 """pull 'key' from parent, if not present"""
5875 return self._Annotated__element.key
5877 @util.memoized_property
5878 def info(self) -> _InfoType:
5879 if TYPE_CHECKING:
5880 assert isinstance(self._Annotated__element, Column)
5881 return self._Annotated__element.info
5883 @util.memoized_property
5884 def _anon_name_label(self) -> str:
5885 return self._Annotated__element._anon_name_label
5888class _truncated_label(quoted_name):
5889 """A unicode subclass used to identify symbolic "
5890 "names that may require truncation."""
5892 __slots__ = ()
5894 def __new__(cls, value: str, quote: Optional[bool] = None) -> Any:
5895 quote = getattr(value, "quote", quote)
5896 # return super(_truncated_label, cls).__new__(cls, value, quote, True)
5897 return super().__new__(cls, value, quote)
5899 def __reduce__(self) -> Any:
5900 return self.__class__, (str(self), self.quote)
5902 def apply_map(self, map_: Mapping[str, Any]) -> str:
5903 return self
5906class conv(_truncated_label):
5907 """Mark a string indicating that a name has already been converted
5908 by a naming convention.
5910 This is a string subclass that indicates a name that should not be
5911 subject to any further naming conventions.
5913 E.g. when we create a :class:`.Constraint` using a naming convention
5914 as follows::
5916 m = MetaData(
5917 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5918 )
5919 t = Table(
5920 "t", m, Column("x", Integer), CheckConstraint("x > 5", name="x5")
5921 )
5923 The name of the above constraint will be rendered as ``"ck_t_x5"``.
5924 That is, the existing name ``x5`` is used in the naming convention as the
5925 ``constraint_name`` token.
5927 In some situations, such as in migration scripts, we may be rendering
5928 the above :class:`.CheckConstraint` with a name that's already been
5929 converted. In order to make sure the name isn't double-modified, the
5930 new name is applied using the :func:`_schema.conv` marker. We can
5931 use this explicitly as follows::
5934 m = MetaData(
5935 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5936 )
5937 t = Table(
5938 "t",
5939 m,
5940 Column("x", Integer),
5941 CheckConstraint("x > 5", name=conv("ck_t_x5")),
5942 )
5944 Where above, the :func:`_schema.conv` marker indicates that the constraint
5945 name here is final, and the name will render as ``"ck_t_x5"`` and not
5946 ``"ck_t_ck_t_x5"``
5948 .. seealso::
5950 :ref:`constraint_naming_conventions`
5952 """
5954 __slots__ = ()
5957# for backwards compatibility in case
5958# someone is re-implementing the
5959# _truncated_identifier() sequence in a custom
5960# compiler
5961_generated_label = _truncated_label
5962_anonymous_label_escape = re.compile(r"[%\(\) \$]+")
5965class _anonymous_label(_truncated_label):
5966 """A unicode subclass used to identify anonymously
5967 generated names."""
5969 __slots__ = ()
5971 @classmethod
5972 def safe_construct_with_key(
5973 cls, seed: int | str, body: str, sanitize_key: bool = False
5974 ) -> typing_Tuple[_anonymous_label, str]:
5975 # need to escape chars that interfere with format
5976 # strings in any case, issue #8724
5977 body = _anonymous_label_escape.sub("_", body)
5979 if sanitize_key:
5980 # sanitize_key is then an extra step used by BindParameter
5981 body = body.strip("_")
5983 key = f"{seed} {body.replace('%', '%%')}"
5984 label = _anonymous_label(f"%({key})s")
5985 return label, key
5987 @classmethod
5988 def safe_construct(
5989 cls, seed: int | str, body: str, sanitize_key: bool = False
5990 ) -> _anonymous_label:
5991 # need to escape chars that interfere with format
5992 # strings in any case, issue #8724
5993 body = _anonymous_label_escape.sub("_", body)
5995 if sanitize_key:
5996 # sanitize_key is then an extra step used by BindParameter
5997 body = body.strip("_")
5999 return _anonymous_label(f"%({seed} {body.replace('%', '%%')})s")
6001 def __add__(self, other: str) -> _anonymous_label:
6002 if "%" in other and not isinstance(other, _anonymous_label):
6003 other = str(other).replace("%", "%%")
6004 else:
6005 other = str(other)
6007 return _anonymous_label(
6008 quoted_name(
6009 str.__add__(self, other),
6010 self.quote,
6011 )
6012 )
6014 def __radd__(self, other: str) -> _anonymous_label:
6015 if "%" in other and not isinstance(other, _anonymous_label):
6016 other = str(other).replace("%", "%%")
6017 else:
6018 other = str(other)
6020 return _anonymous_label(
6021 quoted_name(
6022 str.__add__(other, self),
6023 self.quote,
6024 )
6025 )
6027 def apply_map(self, map_: Mapping[str, Any]) -> str:
6028 if self.quote is not None:
6029 # preserve quoting only if necessary
6030 return quoted_name(self % map_, self.quote)
6031 else:
6032 # else skip the constructor call
6033 return self % map_