Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# sql/elements.py
2# Copyright (C) 2005-2025 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7# mypy: allow-untyped-defs, allow-untyped-calls
9"""Core SQL expression elements, including :class:`_expression.ClauseElement`,
10:class:`_expression.ColumnElement`, and derived classes.
12"""
14from __future__ import annotations
16from decimal import Decimal
17from enum import Enum
18import itertools
19import operator
20import re
21import typing
22from typing import AbstractSet
23from typing import Any
24from typing import Callable
25from typing import cast
26from typing import Dict
27from typing import FrozenSet
28from typing import Generic
29from typing import Iterable
30from typing import Iterator
31from typing import List
32from typing import Literal
33from typing import Mapping
34from typing import Optional
35from typing import overload
36from typing import ParamSpec
37from typing import Sequence
38from typing import Set
39from typing import Tuple as typing_Tuple
40from typing import Type
41from typing import TYPE_CHECKING
42from typing import TypeVar
43from typing import Union
45from . import coercions
46from . import operators
47from . import roles
48from . import traversals
49from . import type_api
50from ._typing import has_schema_attr
51from ._typing import is_named_from_clause
52from ._typing import is_quoted_name
53from ._typing import is_tuple_type
54from .annotation import Annotated
55from .annotation import SupportsWrappingAnnotations
56from .base import _clone
57from .base import _expand_cloned
58from .base import _generative
59from .base import _NoArg
60from .base import Executable
61from .base import 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 Optiona[_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 re-use 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 ]
4533 order_by: Optional[ClauseList] = None
4534 partition_by: Optional[ClauseList] = None
4536 element: ColumnElement[_T]
4537 """The underlying expression object to which this :class:`.Over`
4538 object refers."""
4540 range_: FrameClause | None
4541 rows: FrameClause | None
4542 groups: FrameClause | None
4544 def __init__(
4545 self,
4546 element: ColumnElement[_T],
4547 partition_by: Optional[_ByArgument] = None,
4548 order_by: Optional[_ByArgument] = None,
4549 range_: _FrameIntTuple | FrameClause | None = None,
4550 rows: _FrameIntTuple | FrameClause | None = None,
4551 groups: _FrameIntTuple | FrameClause | None = None,
4552 ):
4553 self.element = element
4554 if order_by is not None:
4555 self.order_by = ClauseList(
4556 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4557 )
4558 if partition_by is not None:
4559 self.partition_by = ClauseList(
4560 *util.to_list(partition_by),
4561 _literal_as_text_role=roles.ByOfRole,
4562 )
4564 if sum(item is not None for item in (range_, rows, groups)) > 1:
4565 raise exc.ArgumentError(
4566 "only one of 'rows', 'range_', or 'groups' may be provided"
4567 )
4568 else:
4569 self.range_ = FrameClause._parse(range_, coerce_int=False)
4570 self.rows = FrameClause._parse(rows, coerce_int=True)
4571 self.groups = FrameClause._parse(groups, coerce_int=True)
4573 if not TYPE_CHECKING:
4575 @util.memoized_property
4576 def type(self) -> TypeEngine[_T]: # noqa: A001
4577 return self.element.type
4579 @util.ro_non_memoized_property
4580 def _from_objects(self) -> List[FromClause]:
4581 return list(
4582 itertools.chain(
4583 *[
4584 c._from_objects
4585 for c in (self.element, self.partition_by, self.order_by)
4586 if c is not None
4587 ]
4588 )
4589 )
4592class FrameClauseType(Enum):
4593 """Frame clause type enum for FrameClause lower_type and upper_type.
4595 .. versionadded:: 2.1
4597 """
4599 UNBOUNDED = 0
4600 """Produces an "UNBOUNDED PRECEDING" or "UNBOUNDED FOLLOWING" frame
4601 clause depending on the position.
4602 Requires a ``None`` value for the corresponding bound value.
4603 """
4604 CURRENT = 1
4605 """Produces a "CURRENT ROW" frame clause.
4606 Requires a ``None`` value for the corresponding bound value.
4607 """
4608 PRECEDING = 2
4609 """Produces a "PRECEDING" frame clause."""
4610 FOLLOWING = 3
4611 """Produces a "FOLLOWING" frame clause."""
4614_require_none = (
4615 FrameClauseType.CURRENT,
4616 FrameClauseType.UNBOUNDED,
4617)
4620class FrameClause(ClauseElement):
4621 """Indicate the 'rows' 'range' or 'group' field of a window function,
4622 e.g. using :class:`.Over`.
4624 .. versionadded:: 2.1
4626 """
4628 __visit_name__ = "frame_clause"
4630 _traverse_internals: _TraverseInternalsType = [
4631 ("lower_bind", InternalTraversal.dp_clauseelement),
4632 ("upper_bind", InternalTraversal.dp_clauseelement),
4633 ("lower_type", InternalTraversal.dp_plain_obj),
4634 ("upper_type", InternalTraversal.dp_plain_obj),
4635 ]
4637 def __init__(
4638 self,
4639 start: Any,
4640 end: Any,
4641 start_frame_type: FrameClauseType,
4642 end_frame_type: FrameClauseType,
4643 _validate: bool = True,
4644 ) -> None:
4645 """Creates a new FrameClause specifying the bounds of a window frame.
4647 :param start: The start value.
4648 :param end: The end value.
4649 :param start_frame_type: The :class:`FrameClauseType` for the
4650 start value.
4651 :param end_frame_type: The :class:`FrameClauseType` for the end value.
4652 """
4653 self.lower_bind = self._as_literal(start)
4654 self.upper_bind = self._as_literal(end)
4655 self.lower_type = FrameClauseType(start_frame_type)
4656 self.upper_type = FrameClauseType(end_frame_type)
4657 if _validate:
4658 if (
4659 self.lower_type in _require_none
4660 and self.lower_bind is not None
4661 ):
4662 raise exc.ArgumentError(
4663 "Cannot specify a value for start with frame type "
4664 f"{self.lower_type.name}"
4665 )
4666 if (
4667 self.upper_type in _require_none
4668 and self.upper_bind is not None
4669 ):
4670 raise exc.ArgumentError(
4671 "Cannot specify a value for end with frame type "
4672 f"{self.upper_type.name}"
4673 )
4675 @classmethod
4676 def _as_literal(cls, value: Any) -> BindParameter[Any] | None:
4677 if value is None:
4678 return None
4679 elif isinstance(value, int):
4680 return literal(value, type_api.INTEGERTYPE)
4681 elif isinstance(value, BindParameter):
4682 return value
4683 else:
4684 return literal(value) # let the default type resolution occur
4686 @classmethod
4687 def _handle_int(
4688 cls, value: Any | None, coerce_int: bool
4689 ) -> tuple[int | None, FrameClauseType]:
4690 if value is None:
4691 return None, FrameClauseType.UNBOUNDED
4693 if coerce_int:
4694 try:
4695 integer = int(value)
4696 except ValueError as err:
4697 raise exc.ArgumentError(
4698 "Integer or None expected for values in rows/groups frame"
4699 ) from err
4700 elif not isinstance(value, int):
4701 raise exc.ArgumentError(
4702 "When using a tuple to specify a range only integer or none "
4703 "values are allowed in the range frame. To specify a "
4704 "different type use the FrameClause directly."
4705 )
4706 else:
4707 integer = value
4708 if integer == 0:
4709 return None, FrameClauseType.CURRENT
4710 elif integer < 0:
4711 return abs(integer), FrameClauseType.PRECEDING
4712 else:
4713 return integer, FrameClauseType.FOLLOWING
4715 @classmethod
4716 def _parse(
4717 cls,
4718 range_: _FrameIntTuple | FrameClause | None,
4719 coerce_int: bool,
4720 ) -> FrameClause | None:
4721 if range_ is None or isinstance(range_, FrameClause):
4722 return range_
4724 try:
4725 r0, r1 = range_
4726 except (ValueError, TypeError) as ve:
4727 raise exc.ArgumentError(
4728 "2-tuple expected for range/rows/groups"
4729 ) from ve
4731 l_b, l_t = cls._handle_int(r0, coerce_int)
4732 u_b, u_t = cls._handle_int(r1, coerce_int)
4734 return FrameClause(
4735 start=l_b,
4736 end=u_b,
4737 start_frame_type=l_t,
4738 end_frame_type=u_t,
4739 _validate=False,
4740 )
4743class AggregateOrderBy(WrapsColumnExpression[_T]):
4744 """Represent an aggregate ORDER BY expression.
4746 This is a special operator against aggregate functions such as
4747 ``array_agg()``, ``json_arrayagg()`` ``string_agg()``, etc. that provides
4748 for an ORDER BY expression, using a syntax that's compatible with
4749 the backend.
4751 :class:`.AggregateOrderBy` is a generalized version of the
4752 :class:`.WithinGroup` construct, the latter of which always provides a
4753 "WITHIN GROUP (ORDER BY ...)" expression. :class:`.AggregateOrderBy` will
4754 also compile to "WITHIN GROUP (ORDER BY ...)" on backends such as Oracle
4755 and SQL Server that don't have another style of aggregate function
4756 ordering.
4758 .. versionadded:: 2.1
4761 """
4763 __visit_name__ = "aggregateorderby"
4765 _traverse_internals: _TraverseInternalsType = [
4766 ("element", InternalTraversal.dp_clauseelement),
4767 ("order_by", InternalTraversal.dp_clauseelement),
4768 ]
4770 order_by: ClauseList
4772 def __init__(
4773 self,
4774 element: Union[FunctionElement[_T], FunctionFilter[_T]],
4775 *order_by: _ColumnExpressionArgument[Any],
4776 ):
4777 self.element = element
4778 if not order_by:
4779 raise TypeError("at least one ORDER BY element is required")
4780 self.order_by = ClauseList(
4781 *util.to_list(order_by), _literal_as_text_role=roles.ByOfRole
4782 )
4784 if not TYPE_CHECKING:
4786 @util.memoized_property
4787 def type(self) -> TypeEngine[_T]: # noqa: A001
4788 return self.element.type
4790 @property
4791 def wrapped_column_expression(self) -> ColumnElement[_T]:
4792 return self.element
4794 def __reduce__(self):
4795 return self.__class__, (self.element,) + (
4796 tuple(self.order_by) if self.order_by is not None else ()
4797 )
4799 def over(
4800 self,
4801 *,
4802 partition_by: _ByArgument | None = None,
4803 order_by: _ByArgument | None = None,
4804 rows: _FrameIntTuple | FrameClause | None = None,
4805 range_: _FrameIntTuple | FrameClause | None = None,
4806 groups: _FrameIntTuple | FrameClause | None = None,
4807 ) -> Over[_T]:
4808 """Produce an OVER clause against this :class:`.WithinGroup`
4809 construct.
4811 This function has the same signature as that of
4812 :meth:`.FunctionElement.over`.
4814 """
4815 return Over(
4816 self,
4817 partition_by=partition_by,
4818 order_by=order_by,
4819 range_=range_,
4820 rows=rows,
4821 groups=groups,
4822 )
4824 @overload
4825 def filter(self) -> Self: ...
4827 @overload
4828 def filter(
4829 self,
4830 __criterion0: _ColumnExpressionArgument[bool],
4831 *criterion: _ColumnExpressionArgument[bool],
4832 ) -> FunctionFilter[_T]: ...
4834 def filter(
4835 self, *criterion: _ColumnExpressionArgument[bool]
4836 ) -> Union[Self, FunctionFilter[_T]]:
4837 """Produce a FILTER clause against this function."""
4838 if not criterion:
4839 return self
4840 return FunctionFilter(self, *criterion)
4842 @util.ro_non_memoized_property
4843 def _from_objects(self) -> List[FromClause]:
4844 return list(
4845 itertools.chain(
4846 *[
4847 c._from_objects
4848 for c in (self.element, self.order_by)
4849 if c is not None
4850 ]
4851 )
4852 )
4855class WithinGroup(AggregateOrderBy[_T]):
4856 """Represent a WITHIN GROUP (ORDER BY) clause.
4858 This is a special operator against so-called
4859 "ordered set aggregate" and "hypothetical
4860 set aggregate" functions, including ``percentile_cont()``,
4861 ``rank()``, ``dense_rank()``, etc.
4863 It's supported only by certain database backends, such as PostgreSQL,
4864 Oracle Database and MS SQL Server.
4866 The :class:`.WithinGroup` construct extracts its type from the
4867 method :meth:`.FunctionElement.within_group_type`. If this returns
4868 ``None``, the function's ``.type`` is used.
4870 """
4872 __visit_name__ = "withingroup"
4873 inherit_cache = True
4875 if not TYPE_CHECKING:
4877 @util.memoized_property
4878 def type(self) -> TypeEngine[_T]: # noqa: A001
4879 wgt = self.element.within_group_type(self)
4880 if wgt is not None:
4881 return wgt
4882 else:
4883 return self.element.type
4886class FunctionFilter(Generative, ColumnElement[_T]):
4887 """Represent a function FILTER clause.
4889 This is a special operator against aggregate and window functions,
4890 which controls which rows are passed to it.
4891 It's supported only by certain database backends.
4893 Invocation of :class:`.FunctionFilter` is via
4894 :meth:`.FunctionElement.filter`::
4896 func.count(1).filter(True)
4898 .. seealso::
4900 :meth:`.FunctionElement.filter`
4902 """
4904 __visit_name__ = "funcfilter"
4906 _traverse_internals: _TraverseInternalsType = [
4907 ("func", InternalTraversal.dp_clauseelement),
4908 ("criterion", InternalTraversal.dp_clauseelement),
4909 ]
4911 criterion: Optional[ColumnElement[bool]] = None
4913 def __init__(
4914 self,
4915 func: Union[FunctionElement[_T], AggregateOrderBy[_T]],
4916 *criterion: _ColumnExpressionArgument[bool],
4917 ):
4918 self.func = func
4919 self.filter.non_generative(self, *criterion) # type: ignore
4921 @_generative
4922 def filter(self, *criterion: _ColumnExpressionArgument[bool]) -> Self:
4923 """Produce an additional FILTER against the function.
4925 This method adds additional criteria to the initial criteria
4926 set up by :meth:`.FunctionElement.filter`.
4928 Multiple criteria are joined together at SQL render time
4929 via ``AND``.
4932 """
4934 for crit in list(criterion):
4935 crit = coercions.expect(roles.WhereHavingRole, crit)
4937 if self.criterion is not None:
4938 self.criterion = self.criterion & crit
4939 else:
4940 self.criterion = crit
4942 return self
4944 def over(
4945 self,
4946 partition_by: _ByArgument | None = None,
4947 order_by: _ByArgument | None = None,
4948 range_: _FrameIntTuple | FrameClause | None = None,
4949 rows: _FrameIntTuple | FrameClause | None = None,
4950 groups: _FrameIntTuple | FrameClause | None = None,
4951 ) -> Over[_T]:
4952 """Produce an OVER clause against this filtered function.
4954 Used against aggregate or so-called "window" functions,
4955 for database backends that support window functions.
4957 The expression::
4959 func.rank().filter(MyClass.y > 5).over(order_by="x")
4961 is shorthand for::
4963 from sqlalchemy import over, funcfilter
4965 over(funcfilter(func.rank(), MyClass.y > 5), order_by="x")
4967 See :func:`_expression.over` for a full description.
4969 """
4970 return Over(
4971 self,
4972 partition_by=partition_by,
4973 order_by=order_by,
4974 range_=range_,
4975 rows=rows,
4976 groups=groups,
4977 )
4979 def within_group(
4980 self, *order_by: _ColumnExpressionArgument[Any]
4981 ) -> WithinGroup[_T]:
4982 """Produce a WITHIN GROUP (ORDER BY expr) clause against
4983 this function.
4984 """
4985 return WithinGroup(self, *order_by)
4987 def within_group_type(
4988 self, within_group: WithinGroup[_T]
4989 ) -> Optional[TypeEngine[_T]]:
4990 return None
4992 def self_group(
4993 self, against: Optional[OperatorType] = None
4994 ) -> Union[Self, Grouping[_T]]:
4995 if operators.is_precedent(operators.filter_op, against):
4996 return Grouping(self)
4997 else:
4998 return self
5000 if not TYPE_CHECKING:
5002 @util.memoized_property
5003 def type(self) -> TypeEngine[_T]: # noqa: A001
5004 return self.func.type
5006 @util.ro_non_memoized_property
5007 def _from_objects(self) -> List[FromClause]:
5008 return list(
5009 itertools.chain(
5010 *[
5011 c._from_objects
5012 for c in (self.func, self.criterion)
5013 if c is not None
5014 ]
5015 )
5016 )
5019class NamedColumn(KeyedColumnElement[_T]):
5020 is_literal = False
5021 table: Optional[FromClause] = None
5022 name: str
5023 key: str
5025 def _compare_name_for_result(self, other):
5026 return (hasattr(other, "name") and self.name == other.name) or (
5027 hasattr(other, "_label") and self._label == other._label
5028 )
5030 @util.ro_memoized_property
5031 def description(self) -> str:
5032 return self.name
5034 @HasMemoized.memoized_attribute
5035 def _tq_key_label(self) -> Optional[str]:
5036 """table qualified label based on column key.
5038 for table-bound columns this is <tablename>_<column key/proxy key>;
5040 all other expressions it resolves to key/proxy key.
5042 """
5043 proxy_key = self._proxy_key
5044 if proxy_key and proxy_key != self.name:
5045 return self._gen_tq_label(proxy_key)
5046 else:
5047 return self._tq_label
5049 @HasMemoized.memoized_attribute
5050 def _tq_label(self) -> Optional[str]:
5051 """table qualified label based on column name.
5053 for table-bound columns this is <tablename>_<columnname>; all other
5054 expressions it resolves to .name.
5056 """
5057 return self._gen_tq_label(self.name)
5059 @HasMemoized.memoized_attribute
5060 def _render_label_in_columns_clause(self):
5061 return True
5063 @HasMemoized.memoized_attribute
5064 def _non_anon_label(self):
5065 return self.name
5067 def _gen_tq_label(
5068 self, name: str, dedupe_on_key: bool = True
5069 ) -> Optional[str]:
5070 return name
5072 def _bind_param(
5073 self,
5074 operator: OperatorType,
5075 obj: Any,
5076 type_: Optional[TypeEngine[_T]] = None,
5077 expanding: bool = False,
5078 ) -> BindParameter[_T]:
5079 return BindParameter(
5080 self.key,
5081 obj,
5082 _compared_to_operator=operator,
5083 _compared_to_type=self.type,
5084 type_=type_,
5085 unique=True,
5086 expanding=expanding,
5087 )
5089 def _make_proxy(
5090 self,
5091 selectable: FromClause,
5092 *,
5093 primary_key: ColumnSet,
5094 foreign_keys: Set[KeyedColumnElement[Any]],
5095 name: Optional[str] = None,
5096 key: Optional[str] = None,
5097 name_is_truncatable: bool = False,
5098 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5099 disallow_is_literal: bool = False,
5100 **kw: Any,
5101 ) -> typing_Tuple[str, ColumnClause[_T]]:
5102 c = ColumnClause(
5103 (
5104 coercions.expect(roles.TruncatedLabelRole, name or self.name)
5105 if name_is_truncatable
5106 else (name or self.name)
5107 ),
5108 type_=self.type,
5109 _selectable=selectable,
5110 is_literal=False,
5111 )
5113 c._propagate_attrs = selectable._propagate_attrs
5114 if name is None:
5115 c.key = self.key
5116 if compound_select_cols:
5117 c._proxies = list(compound_select_cols)
5118 else:
5119 c._proxies = [self]
5121 if selectable._is_clone_of is not None:
5122 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5123 return c.key, c
5126_PS = ParamSpec("_PS")
5129class Label(roles.LabeledColumnExprRole[_T], NamedColumn[_T]):
5130 """Represents a column label (AS).
5132 Represent a label, as typically applied to any column-level
5133 element using the ``AS`` sql keyword.
5135 """
5137 __visit_name__ = "label"
5139 _traverse_internals: _TraverseInternalsType = [
5140 ("name", InternalTraversal.dp_anon_name),
5141 ("type", InternalTraversal.dp_type),
5142 ("_element", InternalTraversal.dp_clauseelement),
5143 ]
5145 _cache_key_traversal = [
5146 ("name", InternalTraversal.dp_anon_name),
5147 ("_element", InternalTraversal.dp_clauseelement),
5148 ]
5150 _element: ColumnElement[_T]
5151 name: str
5153 def __init__(
5154 self,
5155 name: Optional[str],
5156 element: _ColumnExpressionArgument[_T],
5157 type_: Optional[_TypeEngineArgument[_T]] = None,
5158 ):
5159 orig_element = element
5160 element = coercions.expect(
5161 roles.ExpressionElementRole,
5162 element,
5163 apply_propagate_attrs=self,
5164 )
5165 while isinstance(element, Label):
5166 # TODO: this is only covered in test_text.py, but nothing
5167 # fails if it's removed. determine rationale
5168 element = element.element
5170 if name:
5171 self.name = name
5172 else:
5173 self.name = _anonymous_label.safe_construct(
5174 id(self), getattr(element, "name", "anon")
5175 )
5176 if isinstance(orig_element, Label):
5177 # TODO: no coverage for this block, again would be in
5178 # test_text.py where the resolve_label concept is important
5179 self._resolve_label = orig_element._label
5181 self.key = self._tq_label = self._tq_key_label = self.name
5182 self._element = element
5184 self.type = (
5185 type_api.to_instance(type_)
5186 if type_ is not None
5187 else self._element.type
5188 )
5190 self._proxies = [element]
5192 def __reduce__(self):
5193 return self.__class__, (self.name, self._element, self.type)
5195 @HasMemoized.memoized_attribute
5196 def _render_label_in_columns_clause(self):
5197 return True
5199 def _bind_param(self, operator, obj, type_=None, expanding=False):
5200 return BindParameter(
5201 None,
5202 obj,
5203 _compared_to_operator=operator,
5204 type_=type_,
5205 _compared_to_type=self.type,
5206 unique=True,
5207 expanding=expanding,
5208 )
5210 @util.memoized_property
5211 def _is_implicitly_boolean(self):
5212 return self.element._is_implicitly_boolean
5214 @HasMemoized.memoized_attribute
5215 def _allow_label_resolve(self):
5216 return self.element._allow_label_resolve
5218 @property
5219 def _order_by_label_element(self):
5220 return self
5222 def as_reference(self) -> _label_reference[_T]:
5223 """refer to this labeled expression in a clause such as GROUP BY,
5224 ORDER BY etc. as the label name itself, without expanding
5225 into the full expression.
5227 .. versionadded:: 2.1
5229 """
5230 return _label_reference(self)
5232 @HasMemoized.memoized_attribute
5233 def element(self) -> ColumnElement[_T]:
5234 return self._element.self_group(against=operators.as_)
5236 def self_group(self, against: Optional[OperatorType] = None) -> Label[_T]:
5237 return self._apply_to_inner(self._element.self_group, against=against)
5239 def _negate(self):
5240 return self._apply_to_inner(self._element._negate)
5242 def _apply_to_inner(
5243 self,
5244 fn: Callable[_PS, ColumnElement[_T]],
5245 *arg: _PS.args,
5246 **kw: _PS.kwargs,
5247 ) -> Label[_T]:
5248 sub_element = fn(*arg, **kw)
5249 if sub_element is not self._element:
5250 return Label(self.name, sub_element, type_=self.type)
5251 else:
5252 return self
5254 @property
5255 def primary_key(self): # type: ignore[override]
5256 return self.element.primary_key
5258 @property
5259 def foreign_keys(self): # type: ignore[override]
5260 return self.element.foreign_keys
5262 def _copy_internals(
5263 self,
5264 *,
5265 clone: _CloneCallableType = _clone,
5266 anonymize_labels: bool = False,
5267 **kw: Any,
5268 ) -> None:
5269 self._reset_memoizations()
5270 self._element = clone(self._element, **kw)
5271 if anonymize_labels:
5272 self.name = _anonymous_label.safe_construct(
5273 id(self), getattr(self.element, "name", "anon")
5274 )
5275 self.key = self._tq_label = self._tq_key_label = self.name
5277 @util.ro_non_memoized_property
5278 def _from_objects(self) -> List[FromClause]:
5279 return self.element._from_objects
5281 def _make_proxy(
5282 self,
5283 selectable: FromClause,
5284 *,
5285 primary_key: ColumnSet,
5286 foreign_keys: Set[KeyedColumnElement[Any]],
5287 name: Optional[str] = None,
5288 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5289 **kw: Any,
5290 ) -> typing_Tuple[str, ColumnClause[_T]]:
5291 name = self.name if not name else name
5293 key, e = self.element._make_proxy(
5294 selectable,
5295 name=name,
5296 disallow_is_literal=True,
5297 name_is_truncatable=isinstance(name, _truncated_label),
5298 compound_select_cols=compound_select_cols,
5299 primary_key=primary_key,
5300 foreign_keys=foreign_keys,
5301 )
5303 # there was a note here to remove this assertion, which was here
5304 # to determine if we later could support a use case where
5305 # the key and name of a label are separate. But I don't know what
5306 # that case was. For now, this is an unexpected case that occurs
5307 # when a label name conflicts with other columns and select()
5308 # is attempting to disambiguate an explicit label, which is not what
5309 # the user would want. See issue #6090.
5310 if key != self.name and not isinstance(self.name, _anonymous_label):
5311 raise exc.InvalidRequestError(
5312 "Label name %s is being renamed to an anonymous label due "
5313 "to disambiguation "
5314 "which is not supported right now. Please use unique names "
5315 "for explicit labels." % (self.name)
5316 )
5318 e._propagate_attrs = selectable._propagate_attrs
5319 e._proxies.append(self)
5320 if self.type is not None:
5321 e.type = self.type
5323 return self.key, e
5326class ColumnClause(
5327 roles.DDLReferredColumnRole,
5328 roles.LabeledColumnExprRole[_T],
5329 roles.StrAsPlainColumnRole,
5330 Immutable,
5331 NamedColumn[_T],
5332):
5333 """Represents a column expression from any textual string.
5335 The :class:`.ColumnClause`, a lightweight analogue to the
5336 :class:`_schema.Column` class, is typically invoked using the
5337 :func:`_expression.column` function, as in::
5339 from sqlalchemy import column
5341 id, name = column("id"), column("name")
5342 stmt = select(id, name).select_from("user")
5344 The above statement would produce SQL like:
5346 .. sourcecode:: sql
5348 SELECT id, name FROM user
5350 :class:`.ColumnClause` is the immediate superclass of the schema-specific
5351 :class:`_schema.Column` object. While the :class:`_schema.Column`
5352 class has all the
5353 same capabilities as :class:`.ColumnClause`, the :class:`.ColumnClause`
5354 class is usable by itself in those cases where behavioral requirements
5355 are limited to simple SQL expression generation. The object has none of
5356 the associations with schema-level metadata or with execution-time
5357 behavior that :class:`_schema.Column` does,
5358 so in that sense is a "lightweight"
5359 version of :class:`_schema.Column`.
5361 Full details on :class:`.ColumnClause` usage is at
5362 :func:`_expression.column`.
5364 .. seealso::
5366 :func:`_expression.column`
5368 :class:`_schema.Column`
5370 """
5372 table: Optional[FromClause]
5373 is_literal: bool
5375 __visit_name__ = "column"
5377 _traverse_internals: _TraverseInternalsType = [
5378 ("name", InternalTraversal.dp_anon_name),
5379 ("type", InternalTraversal.dp_type),
5380 ("table", InternalTraversal.dp_clauseelement),
5381 ("is_literal", InternalTraversal.dp_boolean),
5382 ]
5384 onupdate: Optional[DefaultGenerator] = None
5385 default: Optional[DefaultGenerator] = None
5386 server_default: Optional[FetchedValue] = None
5387 server_onupdate: Optional[FetchedValue] = None
5389 _is_multiparam_column = False
5391 @property
5392 def _is_star(self): # type: ignore[override]
5393 return self.is_literal and self.name == "*"
5395 def __init__(
5396 self,
5397 text: str,
5398 type_: Optional[_TypeEngineArgument[_T]] = None,
5399 is_literal: bool = False,
5400 _selectable: Optional[FromClause] = None,
5401 ):
5402 self.key = self.name = text
5403 self.table = _selectable
5405 # if type is None, we get NULLTYPE, which is our _T. But I don't
5406 # know how to get the overloads to express that correctly
5407 self.type = type_api.to_instance(type_) # type: ignore
5409 self.is_literal = is_literal
5411 def get_children(self, *, column_tables=False, **kw):
5412 # override base get_children() to not return the Table
5413 # or selectable that is parent to this column. Traversals
5414 # expect the columns of tables and subqueries to be leaf nodes.
5415 return []
5417 @property
5418 def entity_namespace(self):
5419 if self.table is not None:
5420 return self.table.entity_namespace
5421 else:
5422 return super().entity_namespace
5424 def _clone(self, detect_subquery_cols=False, **kw):
5425 if (
5426 detect_subquery_cols
5427 and self.table is not None
5428 and self.table._is_subquery
5429 ):
5430 clone = kw.pop("clone")
5431 table = clone(self.table, **kw)
5432 new = table.c.corresponding_column(self)
5433 return new
5435 return super()._clone(**kw)
5437 @HasMemoized_ro_memoized_attribute
5438 def _from_objects(self) -> List[FromClause]:
5439 t = self.table
5440 if t is not None:
5441 return [t]
5442 else:
5443 return []
5445 @HasMemoized.memoized_attribute
5446 def _render_label_in_columns_clause(self):
5447 return self.table is not None
5449 @property
5450 def _ddl_label(self):
5451 return self._gen_tq_label(self.name, dedupe_on_key=False)
5453 def _compare_name_for_result(self, other):
5454 if (
5455 self.is_literal
5456 or self.table is None
5457 or self.table._is_textual
5458 or not hasattr(other, "proxy_set")
5459 or (
5460 isinstance(other, ColumnClause)
5461 and (
5462 other.is_literal
5463 or other.table is None
5464 or other.table._is_textual
5465 )
5466 )
5467 ):
5468 return (hasattr(other, "name") and self.name == other.name) or (
5469 hasattr(other, "_tq_label")
5470 and self._tq_label == other._tq_label
5471 )
5472 else:
5473 return other.proxy_set.intersection(self.proxy_set)
5475 def _gen_tq_label(
5476 self, name: str, dedupe_on_key: bool = True
5477 ) -> Optional[str]:
5478 """generate table-qualified label
5480 for a table-bound column this is <tablename>_<columnname>.
5482 used primarily for LABEL_STYLE_TABLENAME_PLUS_COL
5483 as well as the .columns collection on a Join object.
5485 """
5486 label: str
5487 t = self.table
5488 if self.is_literal:
5489 return None
5490 elif t is not None and is_named_from_clause(t):
5491 if has_schema_attr(t) and t.schema:
5492 label = (
5493 t.schema.replace(".", "_") + "_" + t.name + ("_" + name)
5494 )
5495 else:
5496 assert not TYPE_CHECKING or isinstance(t, NamedFromClause)
5497 label = t.name + ("_" + name)
5499 # propagate name quoting rules for labels.
5500 if is_quoted_name(name) and name.quote is not None:
5501 if is_quoted_name(label):
5502 label.quote = name.quote
5503 else:
5504 label = quoted_name(label, name.quote)
5505 elif is_quoted_name(t.name) and t.name.quote is not None:
5506 # can't get this situation to occur, so let's
5507 # assert false on it for now
5508 assert not isinstance(label, quoted_name)
5509 label = quoted_name(label, t.name.quote)
5511 if dedupe_on_key:
5512 # ensure the label name doesn't conflict with that of an
5513 # existing column. note that this implies that any Column
5514 # must **not** set up its _label before its parent table has
5515 # all of its other Column objects set up. There are several
5516 # tables in the test suite which will fail otherwise; example:
5517 # table "owner" has columns "name" and "owner_name". Therefore
5518 # column owner.name cannot use the label "owner_name", it has
5519 # to be "owner_name_1".
5520 if label in t.c:
5521 _label = label
5522 counter = 1
5523 while _label in t.c:
5524 _label = label + f"_{counter}"
5525 counter += 1
5526 label = _label
5528 return coercions.expect(roles.TruncatedLabelRole, label)
5530 else:
5531 return name
5533 def _make_proxy(
5534 self,
5535 selectable: FromClause,
5536 *,
5537 primary_key: ColumnSet,
5538 foreign_keys: Set[KeyedColumnElement[Any]],
5539 name: Optional[str] = None,
5540 key: Optional[str] = None,
5541 name_is_truncatable: bool = False,
5542 compound_select_cols: Optional[Sequence[ColumnElement[Any]]] = None,
5543 disallow_is_literal: bool = False,
5544 **kw: Any,
5545 ) -> typing_Tuple[str, ColumnClause[_T]]:
5546 # the "is_literal" flag normally should never be propagated; a proxied
5547 # column is always a SQL identifier and never the actual expression
5548 # being evaluated. however, there is a case where the "is_literal" flag
5549 # might be used to allow the given identifier to have a fixed quoting
5550 # pattern already, so maintain the flag for the proxy unless a
5551 # :class:`.Label` object is creating the proxy. See [ticket:4730].
5552 is_literal = (
5553 not disallow_is_literal
5554 and self.is_literal
5555 and (
5556 # note this does not accommodate for quoted_name differences
5557 # right now
5558 name is None
5559 or name == self.name
5560 )
5561 )
5562 c = self._constructor(
5563 (
5564 coercions.expect(roles.TruncatedLabelRole, name or self.name)
5565 if name_is_truncatable
5566 else (name or self.name)
5567 ),
5568 type_=self.type,
5569 _selectable=selectable,
5570 is_literal=is_literal,
5571 )
5572 c._propagate_attrs = selectable._propagate_attrs
5573 if name is None:
5574 c.key = self.key
5575 if compound_select_cols:
5576 c._proxies = list(compound_select_cols)
5577 else:
5578 c._proxies = [self]
5580 if selectable._is_clone_of is not None:
5581 c._is_clone_of = selectable._is_clone_of.columns.get(c.key)
5582 return c.key, c
5585class TableValuedColumn(NamedColumn[_T]):
5586 __visit_name__ = "table_valued_column"
5588 _traverse_internals: _TraverseInternalsType = [
5589 ("name", InternalTraversal.dp_anon_name),
5590 ("type", InternalTraversal.dp_type),
5591 ("scalar_alias", InternalTraversal.dp_clauseelement),
5592 ]
5594 def __init__(self, scalar_alias: NamedFromClause, type_: TypeEngine[_T]):
5595 self.scalar_alias = scalar_alias
5596 self.key = self.name = scalar_alias.name
5597 self.type = type_
5599 def _copy_internals(
5600 self, clone: _CloneCallableType = _clone, **kw: Any
5601 ) -> None:
5602 self.scalar_alias = clone(self.scalar_alias, **kw)
5603 self.key = self.name = self.scalar_alias.name
5605 @util.ro_non_memoized_property
5606 def _from_objects(self) -> List[FromClause]:
5607 return [self.scalar_alias]
5610class CollationClause(ColumnElement[str]):
5611 __visit_name__ = "collation"
5613 _traverse_internals: _TraverseInternalsType = [
5614 ("collation", InternalTraversal.dp_string)
5615 ]
5617 @classmethod
5618 @util.preload_module("sqlalchemy.sql.sqltypes")
5619 def _create_collation_expression(
5620 cls, expression: _ColumnExpressionArgument[str], collation: str
5621 ) -> BinaryExpression[str]:
5623 sqltypes = util.preloaded.sql_sqltypes
5625 expr = coercions.expect(roles.ExpressionElementRole[str], expression)
5627 if expr.type._type_affinity is sqltypes.String:
5628 collate_type = expr.type._with_collation(collation)
5629 else:
5630 collate_type = expr.type
5632 return BinaryExpression(
5633 expr,
5634 CollationClause(collation),
5635 operators.collate,
5636 type_=collate_type,
5637 )
5639 def __init__(self, collation):
5640 self.collation = collation
5643class _IdentifiedClause(Executable, ClauseElement):
5644 __visit_name__ = "identified"
5646 def __init__(self, ident):
5647 self.ident = ident
5650class SavepointClause(_IdentifiedClause):
5651 __visit_name__ = "savepoint"
5652 inherit_cache = False
5655class RollbackToSavepointClause(_IdentifiedClause):
5656 __visit_name__ = "rollback_to_savepoint"
5657 inherit_cache = False
5660class ReleaseSavepointClause(_IdentifiedClause):
5661 __visit_name__ = "release_savepoint"
5662 inherit_cache = False
5665class quoted_name(util.MemoizedSlots, str):
5666 """Represent a SQL identifier combined with quoting preferences.
5668 :class:`.quoted_name` is a Python unicode/str subclass which
5669 represents a particular identifier name along with a
5670 ``quote`` flag. This ``quote`` flag, when set to
5671 ``True`` or ``False``, overrides automatic quoting behavior
5672 for this identifier in order to either unconditionally quote
5673 or to not quote the name. If left at its default of ``None``,
5674 quoting behavior is applied to the identifier on a per-backend basis
5675 based on an examination of the token itself.
5677 A :class:`.quoted_name` object with ``quote=True`` is also
5678 prevented from being modified in the case of a so-called
5679 "name normalize" option. Certain database backends, such as
5680 Oracle Database, Firebird, and DB2 "normalize" case-insensitive names
5681 as uppercase. The SQLAlchemy dialects for these backends
5682 convert from SQLAlchemy's lower-case-means-insensitive convention
5683 to the upper-case-means-insensitive conventions of those backends.
5684 The ``quote=True`` flag here will prevent this conversion from occurring
5685 to support an identifier that's quoted as all lower case against
5686 such a backend.
5688 The :class:`.quoted_name` object is normally created automatically
5689 when specifying the name for key schema constructs such as
5690 :class:`_schema.Table`, :class:`_schema.Column`, and others.
5691 The class can also be
5692 passed explicitly as the name to any function that receives a name which
5693 can be quoted. Such as to use the :meth:`_engine.Engine.has_table`
5694 method with
5695 an unconditionally quoted name::
5697 from sqlalchemy import create_engine
5698 from sqlalchemy import inspect
5699 from sqlalchemy.sql import quoted_name
5701 engine = create_engine("oracle+oracledb://some_dsn")
5702 print(inspect(engine).has_table(quoted_name("some_table", True)))
5704 The above logic will run the "has table" logic against the Oracle Database
5705 backend, passing the name exactly as ``"some_table"`` without converting to
5706 upper case.
5708 """
5710 __slots__ = "quote", "lower", "upper"
5712 quote: Optional[bool]
5714 @overload
5715 @classmethod
5716 def construct(cls, value: str, quote: Optional[bool]) -> quoted_name: ...
5718 @overload
5719 @classmethod
5720 def construct(cls, value: None, quote: Optional[bool]) -> None: ...
5722 @classmethod
5723 def construct(
5724 cls, value: Optional[str], quote: Optional[bool]
5725 ) -> Optional[quoted_name]:
5726 if value is None:
5727 return None
5728 else:
5729 return quoted_name(value, quote)
5731 def __new__(cls, value: str, quote: Optional[bool]) -> quoted_name:
5732 assert (
5733 value is not None
5734 ), "use quoted_name.construct() for None passthrough"
5735 if isinstance(value, cls) and (quote is None or value.quote == quote):
5736 return value
5737 self = super().__new__(cls, value)
5739 self.quote = quote
5740 return self
5742 def __reduce__(self):
5743 return quoted_name, (str(self), self.quote)
5745 def _memoized_method_lower(self):
5746 if self.quote:
5747 return self
5748 else:
5749 return str(self).lower()
5751 def _memoized_method_upper(self):
5752 if self.quote:
5753 return self
5754 else:
5755 return str(self).upper()
5758def _find_columns(clause: ClauseElement) -> Set[ColumnClause[Any]]:
5759 """locate Column objects within the given expression."""
5761 cols: Set[ColumnClause[Any]] = set()
5762 traverse(clause, {}, {"column": cols.add})
5763 return cols
5766def _type_from_args(args: Sequence[ColumnElement[_T]]) -> TypeEngine[_T]:
5767 for a in args:
5768 if not a.type._isnull:
5769 return a.type
5770 else:
5771 return type_api.NULLTYPE # type: ignore
5774def _corresponding_column_or_error(fromclause, column, require_embedded=False):
5775 c = fromclause.corresponding_column(
5776 column, require_embedded=require_embedded
5777 )
5778 if c is None:
5779 raise exc.InvalidRequestError(
5780 "Given column '%s', attached to table '%s', "
5781 "failed to locate a corresponding column from table '%s'"
5782 % (column, getattr(column, "table", None), fromclause.description)
5783 )
5784 return c
5787class _memoized_property_but_not_nulltype(
5788 util.memoized_property["TypeEngine[_T]"]
5789):
5790 """memoized property, but dont memoize NullType"""
5792 def __get__(self, obj, cls):
5793 if obj is None:
5794 return self
5795 result = self.fget(obj)
5796 if not result._isnull:
5797 obj.__dict__[self.__name__] = result
5798 return result
5801class AnnotatedColumnElement(Annotated):
5802 _Annotated__element: ColumnElement[Any]
5804 def __init__(self, element, values):
5805 Annotated.__init__(self, element, values)
5806 for attr in (
5807 "comparator",
5808 "_proxy_key",
5809 "_tq_key_label",
5810 "_tq_label",
5811 "_non_anon_label",
5812 "type",
5813 ):
5814 self.__dict__.pop(attr, None)
5815 for attr in ("name", "key", "table"):
5816 if self.__dict__.get(attr, False) is None:
5817 self.__dict__.pop(attr)
5819 def _with_annotations(self, values):
5820 clone = super()._with_annotations(values)
5821 for attr in (
5822 "comparator",
5823 "_proxy_key",
5824 "_tq_key_label",
5825 "_tq_label",
5826 "_non_anon_label",
5827 ):
5828 clone.__dict__.pop(attr, None)
5829 return clone
5831 @util.memoized_property
5832 def name(self):
5833 """pull 'name' from parent, if not present"""
5834 return self._Annotated__element.name
5836 @_memoized_property_but_not_nulltype
5837 def type(self):
5838 """pull 'type' from parent and don't cache if null.
5840 type is routinely changed on existing columns within the
5841 mapped_column() initialization process, and "type" is also consulted
5842 during the creation of SQL expressions. Therefore it can change after
5843 it was already retrieved. At the same time we don't want annotated
5844 objects having overhead when expressions are produced, so continue
5845 to memoize, but only when we have a non-null type.
5847 """
5848 return self._Annotated__element.type
5850 @util.memoized_property
5851 def table(self):
5852 """pull 'table' from parent, if not present"""
5853 return self._Annotated__element.table
5855 @util.memoized_property
5856 def key(self):
5857 """pull 'key' from parent, if not present"""
5858 return self._Annotated__element.key
5860 @util.memoized_property
5861 def info(self) -> _InfoType:
5862 if TYPE_CHECKING:
5863 assert isinstance(self._Annotated__element, Column)
5864 return self._Annotated__element.info
5866 @util.memoized_property
5867 def _anon_name_label(self) -> str:
5868 return self._Annotated__element._anon_name_label
5871class _truncated_label(quoted_name):
5872 """A unicode subclass used to identify symbolic "
5873 "names that may require truncation."""
5875 __slots__ = ()
5877 def __new__(cls, value: str, quote: Optional[bool] = None) -> Any:
5878 quote = getattr(value, "quote", quote)
5879 # return super(_truncated_label, cls).__new__(cls, value, quote, True)
5880 return super().__new__(cls, value, quote)
5882 def __reduce__(self) -> Any:
5883 return self.__class__, (str(self), self.quote)
5885 def apply_map(self, map_: Mapping[str, Any]) -> str:
5886 return self
5889class conv(_truncated_label):
5890 """Mark a string indicating that a name has already been converted
5891 by a naming convention.
5893 This is a string subclass that indicates a name that should not be
5894 subject to any further naming conventions.
5896 E.g. when we create a :class:`.Constraint` using a naming convention
5897 as follows::
5899 m = MetaData(
5900 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5901 )
5902 t = Table(
5903 "t", m, Column("x", Integer), CheckConstraint("x > 5", name="x5")
5904 )
5906 The name of the above constraint will be rendered as ``"ck_t_x5"``.
5907 That is, the existing name ``x5`` is used in the naming convention as the
5908 ``constraint_name`` token.
5910 In some situations, such as in migration scripts, we may be rendering
5911 the above :class:`.CheckConstraint` with a name that's already been
5912 converted. In order to make sure the name isn't double-modified, the
5913 new name is applied using the :func:`_schema.conv` marker. We can
5914 use this explicitly as follows::
5917 m = MetaData(
5918 naming_convention={"ck": "ck_%(table_name)s_%(constraint_name)s"}
5919 )
5920 t = Table(
5921 "t",
5922 m,
5923 Column("x", Integer),
5924 CheckConstraint("x > 5", name=conv("ck_t_x5")),
5925 )
5927 Where above, the :func:`_schema.conv` marker indicates that the constraint
5928 name here is final, and the name will render as ``"ck_t_x5"`` and not
5929 ``"ck_t_ck_t_x5"``
5931 .. seealso::
5933 :ref:`constraint_naming_conventions`
5935 """
5937 __slots__ = ()
5940# for backwards compatibility in case
5941# someone is re-implementing the
5942# _truncated_identifier() sequence in a custom
5943# compiler
5944_generated_label = _truncated_label
5945_anonymous_label_escape = re.compile(r"[%\(\) \$]+")
5948class _anonymous_label(_truncated_label):
5949 """A unicode subclass used to identify anonymously
5950 generated names."""
5952 __slots__ = ()
5954 @classmethod
5955 def safe_construct_with_key(
5956 cls, seed: int, body: str, sanitize_key: bool = False
5957 ) -> typing_Tuple[_anonymous_label, str]:
5958 # need to escape chars that interfere with format
5959 # strings in any case, issue #8724
5960 body = _anonymous_label_escape.sub("_", body)
5962 if sanitize_key:
5963 # sanitize_key is then an extra step used by BindParameter
5964 body = body.strip("_")
5966 key = f"{seed} {body.replace('%', '%%')}"
5967 label = _anonymous_label(f"%({key})s")
5968 return label, key
5970 @classmethod
5971 def safe_construct(
5972 cls, seed: int, body: str, sanitize_key: bool = False
5973 ) -> _anonymous_label:
5974 # need to escape chars that interfere with format
5975 # strings in any case, issue #8724
5976 body = _anonymous_label_escape.sub("_", body)
5978 if sanitize_key:
5979 # sanitize_key is then an extra step used by BindParameter
5980 body = body.strip("_")
5982 return _anonymous_label(f"%({seed} {body.replace('%', '%%')})s")
5984 def __add__(self, other: str) -> _anonymous_label:
5985 if "%" in other and not isinstance(other, _anonymous_label):
5986 other = str(other).replace("%", "%%")
5987 else:
5988 other = str(other)
5990 return _anonymous_label(
5991 quoted_name(
5992 str.__add__(self, other),
5993 self.quote,
5994 )
5995 )
5997 def __radd__(self, other: str) -> _anonymous_label:
5998 if "%" in other and not isinstance(other, _anonymous_label):
5999 other = str(other).replace("%", "%%")
6000 else:
6001 other = str(other)
6003 return _anonymous_label(
6004 quoted_name(
6005 str.__add__(other, self),
6006 self.quote,
6007 )
6008 )
6010 def apply_map(self, map_: Mapping[str, Any]) -> str:
6011 if self.quote is not None:
6012 # preserve quoting only if necessary
6013 return quoted_name(self % map_, self.quote)
6014 else:
6015 # else skip the constructor call
6016 return self % map_