Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/sql/base.py: 46%
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/base.py
2# Copyright (C) 2005-2026 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7# mypy: allow-untyped-defs, allow-untyped-calls
9"""Foundational utilities common to many sql modules."""
11from __future__ import annotations
13import collections
14from enum import Enum
15import itertools
16from itertools import zip_longest
17import operator
18import re
19from typing import Any
20from typing import Callable
21from typing import cast
22from typing import Collection
23from typing import Dict
24from typing import Final
25from typing import FrozenSet
26from typing import Generator
27from typing import Generic
28from typing import Iterable
29from typing import Iterator
30from typing import List
31from typing import Mapping
32from typing import MutableMapping
33from typing import NamedTuple
34from typing import NoReturn
35from typing import Optional
36from typing import overload
37from typing import Protocol
38from typing import Sequence
39from typing import Set
40from typing import Tuple
41from typing import Type
42from typing import TYPE_CHECKING
43from typing import TypeGuard
44from typing import TypeVar
45from typing import Union
47from . import roles
48from . import visitors
49from .cache_key import HasCacheKey # noqa
50from .cache_key import MemoizedHasCacheKey # noqa
51from .traversals import HasCopyInternals # noqa
52from .visitors import ClauseVisitor
53from .visitors import ExtendedInternalTraversal
54from .visitors import ExternallyTraversible
55from .visitors import InternalTraversal
56from .. import event
57from .. import exc
58from .. import util
59from ..util import EMPTY_DICT
60from ..util import HasMemoized as HasMemoized
61from ..util import hybridmethod
62from ..util import warn_deprecated
63from ..util.typing import Self
64from ..util.typing import TypeVarTuple
65from ..util.typing import Unpack
67if TYPE_CHECKING:
68 from . import coercions
69 from . import elements
70 from . import type_api
71 from ._orm_types import DMLStrategyArgument
72 from ._orm_types import SynchronizeSessionArgument
73 from ._typing import _CLE
74 from .cache_key import CacheKey
75 from .compiler import SQLCompiler
76 from .dml import Delete
77 from .dml import Insert
78 from .dml import Update
79 from .elements import BindParameter
80 from .elements import ClauseElement
81 from .elements import ClauseList
82 from .elements import ColumnClause # noqa
83 from .elements import ColumnElement
84 from .elements import NamedColumn
85 from .elements import SQLCoreOperations
86 from .elements import TextClause
87 from .schema import Column
88 from .schema import DefaultGenerator
89 from .selectable import _JoinTargetElement
90 from .selectable import _SelectIterable
91 from .selectable import FromClause
92 from .selectable import Select
93 from .visitors import anon_map
94 from ..engine import Connection
95 from ..engine import CursorResult
96 from ..engine.interfaces import _CoreMultiExecuteParams
97 from ..engine.interfaces import _CoreSingleExecuteParams
98 from ..engine.interfaces import _ExecuteOptions
99 from ..engine.interfaces import _ImmutableExecuteOptions
100 from ..engine.interfaces import CacheStats
101 from ..engine.interfaces import Compiled
102 from ..engine.interfaces import CompiledCacheType
103 from ..engine.interfaces import CoreExecuteOptionsParameter
104 from ..engine.interfaces import Dialect
105 from ..engine.interfaces import IsolationLevel
106 from ..engine.interfaces import SchemaTranslateMapType
107 from ..event import dispatcher
109if not TYPE_CHECKING:
110 coercions = None # noqa
111 elements = None # noqa
112 type_api = None # noqa
115_Ts = TypeVarTuple("_Ts")
118class _NoArg(Enum):
119 NO_ARG = 0
121 def __repr__(self):
122 return f"_NoArg.{self.name}"
125NO_ARG: Final = _NoArg.NO_ARG
128class _NoneName(Enum):
129 NONE_NAME = 0
130 """indicate a 'deferred' name that was ultimately the value None."""
133_NONE_NAME: Final = _NoneName.NONE_NAME
135_T = TypeVar("_T", bound=Any)
137_Fn = TypeVar("_Fn", bound=Callable[..., Any])
139_AmbiguousTableNameMap = MutableMapping[str, str]
142class _DefaultDescriptionTuple(NamedTuple):
143 arg: Any
144 is_scalar: Optional[bool]
145 is_callable: Optional[bool]
146 is_sentinel: Optional[bool]
148 @classmethod
149 def _from_column_default(
150 cls, default: Optional[DefaultGenerator]
151 ) -> _DefaultDescriptionTuple:
152 return (
153 _DefaultDescriptionTuple(
154 default.arg, # type: ignore[attr-defined]
155 default.is_scalar,
156 default.is_callable,
157 default.is_sentinel,
158 )
159 if default
160 and (
161 default.has_arg
162 or (not default.for_update and default.is_sentinel)
163 )
164 else _DefaultDescriptionTuple(None, None, None, None)
165 )
168_never_select_column: operator.attrgetter[Any] = operator.attrgetter(
169 "_omit_from_statements"
170)
173class _EntityNamespace(Protocol):
174 def __getattr__(self, key: str) -> SQLCoreOperations[Any]: ...
177class _HasEntityNamespace(Protocol):
178 @util.ro_non_memoized_property
179 def entity_namespace(self) -> _EntityNamespace: ...
182def _is_has_entity_namespace(element: Any) -> TypeGuard[_HasEntityNamespace]:
183 return hasattr(element, "entity_namespace")
186# Remove when https://github.com/python/mypy/issues/14640 will be fixed
187_Self = TypeVar("_Self", bound=Any)
190class Immutable:
191 """mark a ClauseElement as 'immutable' when expressions are cloned.
193 "immutable" objects refers to the "mutability" of an object in the
194 context of SQL DQL and DML generation. Such as, in DQL, one can
195 compose a SELECT or subquery of varied forms, but one cannot modify
196 the structure of a specific table or column within DQL.
197 :class:`.Immutable` is mostly intended to follow this concept, and as
198 such the primary "immutable" objects are :class:`.ColumnClause`,
199 :class:`.Column`, :class:`.TableClause`, :class:`.Table`.
201 """
203 __slots__ = ()
205 _is_immutable: bool = True
207 def unique_params(self, *optionaldict: Any, **kwargs: Any) -> NoReturn:
208 raise NotImplementedError("Immutable objects do not support copying")
210 def params(self, *optionaldict: Any, **kwargs: Any) -> NoReturn:
211 raise NotImplementedError("Immutable objects do not support copying")
213 def _clone(self: _Self, **kw: Any) -> _Self:
214 return self
216 def _copy_internals(
217 self, *, omit_attrs: Iterable[str] = (), **kw: Any
218 ) -> None:
219 pass
222class SingletonConstant(Immutable):
223 """Represent SQL constants like NULL, TRUE, FALSE"""
225 _is_singleton_constant: bool = True
227 _singleton: SingletonConstant
229 def __new__(cls: _T, *arg: Any, **kw: Any) -> _T:
230 return cast(_T, cls._singleton)
232 @util.non_memoized_property
233 def proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
234 raise NotImplementedError()
236 @classmethod
237 def _create_singleton(cls) -> None:
238 obj = object.__new__(cls)
239 obj.__init__() # type: ignore[misc]
241 # for a long time this was an empty frozenset, meaning
242 # a SingletonConstant would never be a "corresponding column" in
243 # a statement. This referred to #6259. However, in #7154 we see
244 # that we do in fact need "correspondence" to work when matching cols
245 # in result sets, so the non-correspondence was moved to a more
246 # specific level when we are actually adapting expressions for SQL
247 # render only.
248 obj.proxy_set = frozenset([obj])
249 cls._singleton = obj
252def _from_objects(
253 *elements: Union[
254 ColumnElement[Any], FromClause, TextClause, _JoinTargetElement
255 ]
256) -> Iterator[FromClause]:
257 return itertools.chain.from_iterable(
258 [element._from_objects for element in elements]
259 )
262def _select_iterables(
263 elements: Iterable[roles.ColumnsClauseRole],
264) -> _SelectIterable:
265 """expand tables into individual columns in the
266 given list of column expressions.
268 """
269 return itertools.chain.from_iterable(
270 [c._select_iterable for c in elements]
271 )
274_SelfGenerativeType = TypeVar("_SelfGenerativeType", bound="_GenerativeType")
277class _GenerativeType(Protocol):
278 def _generate(self) -> Self: ...
281def _generative(fn: _Fn) -> _Fn:
282 """non-caching _generative() decorator.
284 This is basically the legacy decorator that copies the object and
285 runs a method on the new copy.
287 """
289 @util.decorator
290 def _generative(
291 fn: _Fn, self: _SelfGenerativeType, *args: Any, **kw: Any
292 ) -> _SelfGenerativeType:
293 """Mark a method as generative."""
295 self = self._generate()
296 x = fn(self, *args, **kw)
297 assert x is self, "generative methods must return self"
298 return self
300 decorated = _generative(fn)
301 decorated.non_generative = fn # type: ignore[attr-defined]
302 return decorated
305def _exclusive_against(*names: str, **kw: Any) -> Callable[[_Fn], _Fn]:
306 msgs: Dict[str, str] = kw.pop("msgs", {})
308 defaults: Dict[str, str] = kw.pop("defaults", {})
310 getters: List[Tuple[str, operator.attrgetter[Any], Optional[str]]] = [
311 (name, operator.attrgetter(name), defaults.get(name, None))
312 for name in names
313 ]
315 @util.decorator
316 def check(fn: _Fn, *args: Any, **kw: Any) -> Any:
317 # make pylance happy by not including "self" in the argument
318 # list
319 self = args[0]
320 args = args[1:]
321 for name, getter, default_ in getters:
322 if getter(self) is not default_:
323 msg = msgs.get(
324 name,
325 "Method %s() has already been invoked on this %s construct"
326 % (fn.__name__, self.__class__),
327 )
328 raise exc.InvalidRequestError(msg)
329 return fn(self, *args, **kw)
331 return check
334def _clone(element, **kw):
335 return element._clone(**kw)
338def _expand_cloned(
339 elements: Iterable[_CLE],
340) -> Iterable[_CLE]:
341 """expand the given set of ClauseElements to be the set of all 'cloned'
342 predecessors.
344 """
345 # TODO: cython candidate
346 return itertools.chain(*[x._cloned_set for x in elements])
349def _de_clone(
350 elements: Iterable[_CLE],
351) -> Iterable[_CLE]:
352 for x in elements:
353 while x._is_clone_of is not None:
354 x = x._is_clone_of
355 yield x
358def _cloned_intersection(a: Iterable[_CLE], b: Iterable[_CLE]) -> Set[_CLE]:
359 """return the intersection of sets a and b, counting
360 any overlap between 'cloned' predecessors.
362 The returned set is in terms of the entities present within 'a'.
364 """
365 all_overlap: Set[_CLE] = set(_expand_cloned(a)).intersection(
366 _expand_cloned(b)
367 )
368 return {elem for elem in a if all_overlap.intersection(elem._cloned_set)}
371def _cloned_difference(a: Iterable[_CLE], b: Iterable[_CLE]) -> Set[_CLE]:
372 all_overlap: Set[_CLE] = set(_expand_cloned(a)).intersection(
373 _expand_cloned(b)
374 )
375 return {
376 elem for elem in a if not all_overlap.intersection(elem._cloned_set)
377 }
380class DialectKWArgConst(Enum):
381 """Constants for dialect argument defaults in
382 :attr:`.DefaultDialect.construct_arguments`.
384 """
386 REFLECTED_ONLY = 1
387 """Mark a dialect argument as database state reported by reflection.
389 Applies to any construct that takes part in
390 :attr:`.DefaultDialect.construct_arguments`, such as :class:`.Table`,
391 :class:`.Column`, :class:`.Index` or :class:`.CheckConstraint`. The
392 value is kept in the construct's read-only
393 :attr:`.DialectKWArgs.reflect_only_elements` mapping, separate from the
394 options used to generate DDL.
396 .. seealso::
398 :attr:`.DialectKWArgs.reflect_only_elements`
400 .. versionadded:: 2.1
402 """
405class _DialectArgView(MutableMapping[str, Any]):
406 """A dictionary view of dialect-level arguments in the form
407 <dialectname>_<argument_name>.
409 """
411 __slots__ = ("obj",)
413 def __init__(self, obj: DialectKWArgs) -> None:
414 self.obj = obj
416 def _key(self, key: str) -> Tuple[str, str]:
417 try:
418 dialect, value_key = key.split("_", 1)
419 except ValueError as err:
420 raise KeyError(key) from err
421 else:
422 return dialect, value_key
424 def __getitem__(self, key: str) -> Any:
425 dialect, value_key = self._key(key)
427 try:
428 opt = self.obj.dialect_options[dialect]
429 except exc.NoSuchModuleError as err:
430 raise KeyError(key) from err
431 else:
432 return opt[value_key]
434 def __setitem__(self, key: str, value: Any) -> None:
435 try:
436 dialect, value_key = self._key(key)
437 except KeyError as err:
438 raise exc.ArgumentError(
439 "Keys must be of the form <dialectname>_<argname>"
440 ) from err
441 else:
442 self.obj.dialect_options[dialect][value_key] = value
444 def __delitem__(self, key: str) -> None:
445 dialect, value_key = self._key(key)
446 del self.obj.dialect_options[dialect][value_key]
448 def __len__(self) -> int:
449 return sum(
450 len(args._non_defaults)
451 for args in self.obj.dialect_options.values()
452 )
454 def __iter__(self) -> Generator[str, None, None]:
455 return (
456 "%s_%s" % (dialect_name, value_name)
457 for dialect_name in self.obj.dialect_options
458 for value_name in self.obj.dialect_options[
459 dialect_name
460 ]._non_defaults
461 )
464class _ReflectOnlyView(Mapping[str, Mapping[str, Any]]):
465 """A read-only dictionary view of reflection-only dialect-level
466 arguments, keyed to <dialectname>, then <argument_name>.
468 A dialect name that has no reflection-only arguments present returns
469 an empty mapping.
471 """
473 __slots__ = ("obj",)
475 def __init__(self, obj: DialectKWArgs) -> None:
476 self.obj = obj
478 def __getitem__(self, key: str) -> Mapping[str, Any]:
479 # check membership first so that the lookup does not attempt
480 # to load a dialect
481 if key in self.obj.dialect_options:
482 return self.obj.dialect_options[key]._reflect_only_elements
483 else:
484 return EMPTY_DICT
486 def __contains__(self, key: object) -> bool:
487 return bool(
488 isinstance(key, str)
489 and key in self.obj.dialect_options
490 and self.obj.dialect_options[key]._reflect_only_elements
491 )
493 def __len__(self) -> int:
494 return sum(1 for _ in self)
496 def __iter__(self) -> Generator[str, None, None]:
497 return (
498 dialect_name
499 for dialect_name, args in self.obj.dialect_options.items()
500 if args._reflect_only_elements
501 )
504class _DialectArgDict(MutableMapping[str, Any]):
505 """A dictionary view of dialect-level arguments for a specific
506 dialect.
508 Maintains a separate collection of user-specified arguments
509 and dialect-specified default arguments.
511 """
513 def __init__(self) -> None:
514 self._non_defaults: Dict[str, Any] = {}
515 self._defaults: Dict[str, Any] = {}
516 self._reflection_only_keys: FrozenSet[str] = util.EMPTY_SET
517 self._reflect_only_elements: util.immutabledict[str, Any] = (
518 util.EMPTY_DICT
519 )
521 def __len__(self) -> int:
522 return len(set(self._non_defaults).union(self._defaults))
524 def __iter__(self) -> Iterator[str]:
525 return iter(set(self._non_defaults).union(self._defaults))
527 def __getitem__(self, key: str) -> Any:
528 if key in self._non_defaults:
529 return self._non_defaults[key]
530 else:
531 return self._defaults[key]
533 def __setitem__(self, key: str, value: Any) -> None:
534 if key in self._reflection_only_keys:
535 self._reflect_only_elements = self._reflect_only_elements.union(
536 {key: value}
537 )
538 else:
539 self._non_defaults[key] = value
541 def __delitem__(self, key: str) -> None:
542 del self._non_defaults[key]
545@util.preload_module("sqlalchemy.dialects")
546def _kw_reg_for_dialect(dialect_name: str) -> Optional[Dict[Any, Any]]:
547 dialect_cls = util.preloaded.dialects.registry.load(dialect_name)
548 if dialect_cls.construct_arguments is None:
549 return None
550 return dict(dialect_cls.construct_arguments)
553class DialectKWArgs:
554 """Establish the ability for a class to have dialect-specific arguments
555 with defaults and constructor validation.
557 The :class:`.DialectKWArgs` interacts with the
558 :attr:`.DefaultDialect.construct_arguments` present on a dialect.
560 .. seealso::
562 :attr:`.DefaultDialect.construct_arguments`
564 """
566 __slots__ = ()
568 _dialect_kwargs_traverse_internals: List[Tuple[str, Any]] = [
569 ("dialect_options", InternalTraversal.dp_dialect_options)
570 ]
572 def get_dialect_option(
573 self,
574 dialect: Dialect,
575 argument_name: str,
576 *,
577 else_: Any = None,
578 deprecated_fallback: Optional[str] = None,
579 ) -> Any:
580 r"""Return the value of a dialect-specific option, or *else_* if
581 this dialect does not register the given argument.
583 This is useful for DDL compilers that may be inherited by
584 third-party dialects whose ``construct_arguments`` do not
585 include the same set of keys as the parent dialect.
587 :param dialect: The dialect for which to retrieve the option.
588 :param argument_name: The name of the argument to retrieve.
589 :param else\_: The value to return if the argument is not present.
590 :param deprecated_fallback: Optional dialect name to fall back to
591 if the argument is not present for the current dialect. If the
592 argument is present for the fallback dialect but not the current
593 dialect, a deprecation warning will be emitted.
595 """
597 registry = DialectKWArgs._kw_registry[dialect.name]
598 if registry is None:
599 return else_
601 if argument_name in registry.get(self.__class__, {}):
602 if (
603 argument_name
604 in self.dialect_options[dialect.name]._reflection_only_keys
605 ):
606 return else_
607 if (
608 deprecated_fallback is None
609 or dialect.name == deprecated_fallback
610 ):
611 return self.dialect_options[dialect.name][argument_name]
613 # deprecated_fallback is present; need to look in two places
615 # Current dialect has this option registered.
616 # Check if user explicitly set it.
617 if (
618 dialect.name in self.dialect_options
619 and argument_name
620 in self.dialect_options[dialect.name]._non_defaults
621 ):
622 # User explicitly set this dialect's option - use it
623 return self.dialect_options[dialect.name][argument_name]
625 # User didn't set current dialect's option.
626 # Check for deprecated fallback.
627 elif (
628 deprecated_fallback in self.dialect_options
629 and argument_name
630 in self.dialect_options[deprecated_fallback]._non_defaults
631 ):
632 # User set fallback option but not current dialect's option
633 warn_deprecated(
634 f"Using '{deprecated_fallback}_{argument_name}' "
635 f"with the '{dialect.name}' dialect is deprecated; "
636 f"please additionally specify "
637 f"'{dialect.name}_{argument_name}'.",
638 version="2.1",
639 )
640 return self.dialect_options[deprecated_fallback][argument_name]
642 # Return default value
643 return self.dialect_options[dialect.name][argument_name]
644 else:
645 # Current dialect doesn't have the option registered at all.
646 # Don't warn - if a third-party dialect doesn't support an
647 # option, that's their choice, not a deprecation case.
648 return else_
650 @classmethod
651 def argument_for(
652 cls, dialect_name: str, argument_name: str, default: Any
653 ) -> None:
654 """Add a new kind of dialect-specific keyword argument for this class.
656 E.g.::
658 Index.argument_for("mydialect", "length", None)
660 some_index = Index("a", "b", mydialect_length=5)
662 The :meth:`.DialectKWArgs.argument_for` method is a per-argument
663 way adding extra arguments to the
664 :attr:`.DefaultDialect.construct_arguments` dictionary. This
665 dictionary provides a list of argument names accepted by various
666 schema-level constructs on behalf of a dialect.
668 New dialects should typically specify this dictionary all at once as a
669 data member of the dialect class. The use case for ad-hoc addition of
670 argument names is typically for end-user code that is also using
671 a custom compilation scheme which consumes the additional arguments.
673 :param dialect_name: name of a dialect. The dialect must be
674 locatable, else a :class:`.NoSuchModuleError` is raised. The
675 dialect must also include an existing
676 :attr:`.DefaultDialect.construct_arguments` collection, indicating
677 that it participates in the keyword-argument validation and default
678 system, else :class:`.ArgumentError` is raised. If the dialect does
679 not include this collection, then any keyword argument can be
680 specified on behalf of this dialect already. All dialects packaged
681 within SQLAlchemy include this collection, however for third party
682 dialects, support may vary.
684 :param argument_name: name of the parameter.
686 :param default: default value of the parameter.
688 """
690 construct_arg_dictionary: Optional[Dict[Any, Any]] = (
691 DialectKWArgs._kw_registry[dialect_name]
692 )
693 if construct_arg_dictionary is None:
694 raise exc.ArgumentError(
695 "Dialect '%s' does have keyword-argument "
696 "validation and defaults enabled configured" % dialect_name
697 )
698 if cls not in construct_arg_dictionary:
699 construct_arg_dictionary[cls] = {}
700 construct_arg_dictionary[cls][argument_name] = default
702 @property
703 def dialect_kwargs(self) -> _DialectArgView:
704 """A collection of keyword arguments specified as dialect-specific
705 options to this construct.
707 The arguments are present here in their original ``<dialect>_<kwarg>``
708 format. Only arguments that were actually passed are included;
709 unlike the :attr:`.DialectKWArgs.dialect_options` collection, which
710 contains all options known by this dialect including defaults.
712 The collection is also writable; keys are accepted of the
713 form ``<dialect>_<kwarg>`` where the value will be assembled
714 into the list of options.
716 .. seealso::
718 :attr:`.DialectKWArgs.dialect_options` - nested dictionary form
720 """
721 return _DialectArgView(self)
723 @property
724 def kwargs(self) -> _DialectArgView:
725 """A synonym for :attr:`.DialectKWArgs.dialect_kwargs`."""
726 return self.dialect_kwargs
728 @property
729 def reflect_only_elements(self) -> Mapping[str, Mapping[str, Any]]:
730 """A read-only collection of dialect-specific database state
731 reported by reflection, separate from DDL options.
733 Holds values for arguments whose
734 :attr:`.DefaultDialect.construct_arguments` default is
735 :attr:`.DialectKWArgConst.REFLECTED_ONLY`. Like
736 :attr:`.DialectKWArgs.dialect_options`, this is a two-level nested
737 collection keyed to ``<dialect_name>`` and ``<argument_name>``; a
738 dialect name with no values present returns an empty mapping, for
739 example::
741 invalid = my_index.reflect_only_elements["postgresql"].get("invalid")
743 These values are not included in
744 :attr:`.DialectKWArgs.dialect_options` or
745 :attr:`.DialectKWArgs.dialect_kwargs` and take no part in DDL
746 compilation. They are carried along to copies of the object, such
747 as those produced by :meth:`.Table.to_metadata`.
749 .. versionadded:: 2.1
751 .. seealso::
753 :attr:`.DialectKWArgConst.REFLECTED_ONLY`
755 """ # noqa: E501
756 return _ReflectOnlyView(self)
758 @property
759 def _reflect_only_kwargs(self) -> Dict[str, Any]:
760 """The contents of :attr:`.DialectKWArgs.reflect_only_elements` in
761 flat ``<dialect>_<argument>`` form, suitable to be passed to the
762 constructor of a copy of this object."""
764 return {
765 f"{dialect_name}_{key}": value
766 for dialect_name, elements in self.reflect_only_elements.items()
767 for key, value in elements.items()
768 }
770 def _copy_reflect_only_elements(self, other: DialectKWArgs) -> None:
771 """Copy the contents of :attr:`.DialectKWArgs.reflect_only_elements`
772 onto ``other``, a copy of this object."""
774 if "dialect_options" not in self.__dict__:
775 return
776 for dialect_name, args in self.dialect_options.items():
777 if args._reflect_only_elements:
778 other.dialect_options[dialect_name]._reflect_only_elements = (
779 args._reflect_only_elements
780 )
782 _kw_registry: util.PopulateDict[str, Optional[Dict[Any, Any]]] = (
783 util.PopulateDict(_kw_reg_for_dialect)
784 )
786 @classmethod
787 def _kw_reg_for_dialect_cls(cls, dialect_name: str) -> _DialectArgDict:
788 construct_arg_dictionary = DialectKWArgs._kw_registry[dialect_name]
789 d = _DialectArgDict()
791 if construct_arg_dictionary is None:
792 d._defaults.update({"*": None})
793 else:
794 for cls in reversed(cls.__mro__):
795 if cls in construct_arg_dictionary:
796 d._defaults.update(construct_arg_dictionary[cls])
797 d._reflection_only_keys = frozenset(
798 key
799 for key, value in d._defaults.items()
800 if value is DialectKWArgConst.REFLECTED_ONLY
801 )
802 for key in d._reflection_only_keys:
803 del d._defaults[key]
804 return d
806 @util.memoized_property
807 def dialect_options(self) -> util.PopulateDict[str, _DialectArgDict]:
808 """A collection of keyword arguments specified as dialect-specific
809 options to this construct.
811 This is a two-level nested registry, keyed to ``<dialect_name>``
812 and ``<argument_name>``. For example, the ``postgresql_where``
813 argument would be locatable as::
815 arg = my_object.dialect_options["postgresql"]["where"]
817 .. versionadded:: 0.9.2
819 Arguments a dialect declares as
820 :attr:`.DialectKWArgConst.REFLECTED_ONLY` are not included in this
821 collection; they are available from
822 :attr:`.DialectKWArgs.reflect_only_elements`.
824 .. seealso::
826 :attr:`.DialectKWArgs.dialect_kwargs` - flat dictionary form
828 :attr:`.DialectKWArgs.reflect_only_elements` - database state
829 reported by reflection
831 """
833 return util.PopulateDict(self._kw_reg_for_dialect_cls)
835 def _validate_dialect_kwargs(self, kwargs: Dict[str, Any]) -> None:
836 # validate remaining kwargs that they all specify DB prefixes
838 if not kwargs:
839 return
841 for k in kwargs:
842 m = re.match("^(.+?)_(.+)$", k)
843 if not m:
844 raise TypeError(
845 "Additional arguments should be "
846 "named <dialectname>_<argument>, got '%s'" % k
847 )
848 dialect_name, arg_name = m.group(1, 2)
850 try:
851 construct_arg_dictionary = self.dialect_options[dialect_name]
852 except exc.NoSuchModuleError:
853 util.warn(
854 "Can't validate argument %r; can't "
855 "locate any SQLAlchemy dialect named %r"
856 % (k, dialect_name)
857 )
858 self.dialect_options[dialect_name] = d = _DialectArgDict()
859 d._defaults.update({"*": None})
860 d._non_defaults[arg_name] = kwargs[k]
861 else:
862 if (
863 "*" not in construct_arg_dictionary
864 and arg_name not in construct_arg_dictionary
865 and arg_name
866 not in construct_arg_dictionary._reflection_only_keys
867 ):
868 raise exc.ArgumentError(
869 "Argument %r is not accepted by "
870 "dialect %r on behalf of %r"
871 % (k, dialect_name, self.__class__)
872 )
873 else:
874 construct_arg_dictionary[arg_name] = kwargs[k]
877class CompileState:
878 """Produces additional object state necessary for a statement to be
879 compiled.
881 the :class:`.CompileState` class is at the base of classes that assemble
882 state for a particular statement object that is then used by the
883 compiler. This process is essentially an extension of the process that
884 the SQLCompiler.visit_XYZ() method takes, however there is an emphasis
885 on converting raw user intent into more organized structures rather than
886 producing string output. The top-level :class:`.CompileState` for the
887 statement being executed is also accessible when the execution context
888 works with invoking the statement and collecting results.
890 The production of :class:`.CompileState` is specific to the compiler, such
891 as within the :meth:`.SQLCompiler.visit_insert`,
892 :meth:`.SQLCompiler.visit_select` etc. methods. These methods are also
893 responsible for associating the :class:`.CompileState` with the
894 :class:`.SQLCompiler` itself, if the statement is the "toplevel" statement,
895 i.e. the outermost SQL statement that's actually being executed.
896 There can be other :class:`.CompileState` objects that are not the
897 toplevel, such as when a SELECT subquery or CTE-nested
898 INSERT/UPDATE/DELETE is generated.
900 .. versionadded:: 1.4
902 """
904 __slots__ = ("statement", "_ambiguous_table_name_map")
906 plugins: Dict[Tuple[str, str], Type[CompileState]] = {}
908 _ambiguous_table_name_map: Optional[_AmbiguousTableNameMap]
910 @classmethod
911 def create_for_statement(
912 cls, statement: Executable, compiler: SQLCompiler, **kw: Any
913 ) -> CompileState:
914 # factory construction.
916 if statement._propagate_attrs:
917 plugin_name = statement._propagate_attrs.get(
918 "compile_state_plugin", "default"
919 )
920 klass = cls.plugins.get(
921 (plugin_name, statement._effective_plugin_target), None
922 )
923 if klass is None:
924 klass = cls.plugins[
925 ("default", statement._effective_plugin_target)
926 ]
928 else:
929 klass = cls.plugins[
930 ("default", statement._effective_plugin_target)
931 ]
933 if klass is cls:
934 return cls(statement, compiler, **kw)
935 else:
936 return klass.create_for_statement(statement, compiler, **kw)
938 def __init__(self, statement, compiler, **kw):
939 self.statement = statement
941 @classmethod
942 def get_plugin_class(
943 cls, statement: Executable
944 ) -> Optional[Type[CompileState]]:
945 plugin_name = statement._propagate_attrs.get(
946 "compile_state_plugin", None
947 )
949 if plugin_name:
950 key = (plugin_name, statement._effective_plugin_target)
951 if key in cls.plugins:
952 return cls.plugins[key]
954 # there's no case where we call upon get_plugin_class() and want
955 # to get None back, there should always be a default. return that
956 # if there was no plugin-specific class (e.g. Insert with "orm"
957 # plugin)
958 try:
959 return cls.plugins[("default", statement._effective_plugin_target)]
960 except KeyError:
961 return None
963 @classmethod
964 def _get_plugin_class_for_plugin(
965 cls, statement: Executable, plugin_name: str
966 ) -> Optional[Type[CompileState]]:
967 try:
968 return cls.plugins[
969 (plugin_name, statement._effective_plugin_target)
970 ]
971 except KeyError:
972 return None
974 @classmethod
975 def plugin_for(
976 cls, plugin_name: str, visit_name: str
977 ) -> Callable[[_Fn], _Fn]:
978 def decorate(cls_to_decorate):
979 cls.plugins[(plugin_name, visit_name)] = cls_to_decorate
980 return cls_to_decorate
982 return decorate
985class Generative(HasMemoized):
986 """Provide a method-chaining pattern in conjunction with the
987 @_generative decorator."""
989 def _generate(self) -> Self:
990 skip = self._memoized_keys
991 cls = self.__class__
992 s = cls.__new__(cls)
993 if skip:
994 # ensure this iteration remains atomic
995 s.__dict__ = {
996 k: v for k, v in self.__dict__.copy().items() if k not in skip
997 }
998 else:
999 s.__dict__ = self.__dict__.copy()
1000 return s
1003class InPlaceGenerative(HasMemoized):
1004 """Provide a method-chaining pattern in conjunction with the
1005 @_generative decorator that mutates in place."""
1007 __slots__ = ()
1009 def _generate(self) -> Self:
1010 skip = self._memoized_keys
1011 # note __dict__ needs to be in __slots__ if this is used
1012 for k in skip:
1013 self.__dict__.pop(k, None)
1014 return self
1017class HasCompileState(Generative):
1018 """A class that has a :class:`.CompileState` associated with it."""
1020 _compile_state_plugin: Optional[Type[CompileState]] = None
1022 _attributes: util.immutabledict[str, Any] = util.EMPTY_DICT
1024 _compile_state_factory = CompileState.create_for_statement
1027class _MetaOptions(type):
1028 """metaclass for the Options class.
1030 This metaclass is actually necessary despite the availability of the
1031 ``__init_subclass__()`` hook as this type also provides custom class-level
1032 behavior for the ``__add__()`` method.
1034 """
1036 _cache_attrs: Tuple[str, ...]
1038 def __add__(self, other):
1039 o1 = self()
1041 if set(other).difference(self._cache_attrs):
1042 raise TypeError(
1043 "dictionary contains attributes not covered by "
1044 "Options class %s: %r"
1045 % (self, set(other).difference(self._cache_attrs))
1046 )
1048 o1.__dict__.update(other)
1049 return o1
1051 if TYPE_CHECKING:
1053 def __getattr__(self, key: str) -> Any: ...
1055 def __setattr__(self, key: str, value: Any) -> None: ...
1057 def __delattr__(self, key: str) -> None: ...
1060class Options(metaclass=_MetaOptions):
1061 """A cacheable option dictionary with defaults."""
1063 __slots__ = ()
1065 _cache_attrs: Tuple[str, ...]
1067 def __init_subclass__(cls) -> None:
1068 dict_ = cls.__dict__
1069 cls._cache_attrs = tuple(
1070 sorted(
1071 d
1072 for d in dict_
1073 if not d.startswith("__")
1074 and d not in ("_cache_key_traversal",)
1075 )
1076 )
1077 super().__init_subclass__()
1079 def __init__(self, **kw: Any) -> None:
1080 self.__dict__.update(kw)
1082 def __add__(self, other):
1083 o1 = self.__class__.__new__(self.__class__)
1084 o1.__dict__.update(self.__dict__)
1086 if set(other).difference(self._cache_attrs):
1087 raise TypeError(
1088 "dictionary contains attributes not covered by "
1089 "Options class %s: %r"
1090 % (self, set(other).difference(self._cache_attrs))
1091 )
1093 o1.__dict__.update(other)
1094 return o1
1096 def __eq__(self, other):
1097 # TODO: very inefficient. This is used only in test suites
1098 # right now.
1099 for a, b in zip_longest(self._cache_attrs, other._cache_attrs):
1100 if getattr(self, a) != getattr(other, b):
1101 return False
1102 return True
1104 def __repr__(self) -> str:
1105 # TODO: fairly inefficient, used only in debugging right now.
1107 return "%s(%s)" % (
1108 self.__class__.__name__,
1109 ", ".join(
1110 "%s=%r" % (k, self.__dict__[k])
1111 for k in self._cache_attrs
1112 if k in self.__dict__
1113 ),
1114 )
1116 @classmethod
1117 def isinstance(cls, klass: Type[Any]) -> bool:
1118 return issubclass(cls, klass)
1120 @hybridmethod
1121 def add_to_element(self, name: str, value: str) -> Any:
1122 return self + {name: getattr(self, name) + value}
1124 @hybridmethod
1125 def _state_dict_inst(self) -> Mapping[str, Any]:
1126 return self.__dict__
1128 _state_dict_const: util.immutabledict[str, Any] = util.EMPTY_DICT
1130 @_state_dict_inst.classlevel
1131 def _state_dict(cls) -> Mapping[str, Any]:
1132 return cls._state_dict_const
1134 @classmethod
1135 def safe_merge(cls, other: "Options") -> Any:
1136 d = other._state_dict()
1138 # only support a merge with another object of our class
1139 # and which does not have attrs that we don't. otherwise
1140 # we risk having state that might not be part of our cache
1141 # key strategy
1143 if (
1144 cls is not other.__class__
1145 and other._cache_attrs
1146 and set(other._cache_attrs).difference(cls._cache_attrs)
1147 ):
1148 raise TypeError(
1149 "other element %r is not empty, is not of type %s, "
1150 "and contains attributes not covered here %r"
1151 % (
1152 other,
1153 cls,
1154 set(other._cache_attrs).difference(cls._cache_attrs),
1155 )
1156 )
1157 return cls + d
1159 @classmethod
1160 def from_execution_options(
1161 cls,
1162 key: str,
1163 attrs: set[str],
1164 exec_options: Mapping[str, Any],
1165 statement_exec_options: Mapping[str, Any],
1166 ) -> Tuple["Options", Mapping[str, Any]]:
1167 """process Options argument in terms of execution options.
1170 e.g.::
1172 (
1173 load_options,
1174 execution_options,
1175 ) = QueryContext.default_load_options.from_execution_options(
1176 "_sa_orm_load_options",
1177 {"populate_existing", "autoflush", "yield_per"},
1178 execution_options,
1179 statement._execution_options,
1180 )
1182 get back the Options and refresh "_sa_orm_load_options" in the
1183 exec options dict w/ the Options as well
1185 """
1187 # common case is that no options we are looking for are
1188 # in either dictionary, so cancel for that first
1189 check_argnames = attrs.intersection(
1190 set(exec_options).union(statement_exec_options)
1191 )
1193 existing_options = exec_options.get(key, cls)
1195 if check_argnames:
1196 result = {}
1197 for argname in check_argnames:
1198 local = "_" + argname
1199 if argname in exec_options:
1200 result[local] = exec_options[argname]
1201 elif argname in statement_exec_options:
1202 result[local] = statement_exec_options[argname]
1204 new_options = existing_options + result
1205 exec_options = util.EMPTY_DICT.merge_with(
1206 exec_options, {key: new_options}
1207 )
1208 return new_options, exec_options
1210 else:
1211 return existing_options, exec_options
1213 if TYPE_CHECKING:
1215 def __getattr__(self, key: str) -> Any: ...
1217 def __setattr__(self, key: str, value: Any) -> None: ...
1219 def __delattr__(self, key: str) -> None: ...
1222class CacheableOptions(Options, HasCacheKey):
1223 __slots__ = ()
1225 @hybridmethod
1226 def _gen_cache_key_inst(
1227 self, anon_map: Any, bindparams: List[BindParameter[Any]]
1228 ) -> Optional[Tuple[Any]]:
1229 # _gen_cache_key is a compiled function in _cache_key_cy; its
1230 # cython directives make mypy see it as untyped
1231 return HasCacheKey._gen_cache_key( # type: ignore[no-any-return] # noqa: E501
1232 self, anon_map, bindparams
1233 )
1235 @_gen_cache_key_inst.classlevel
1236 def _gen_cache_key(
1237 cls, anon_map: "anon_map", bindparams: List[BindParameter[Any]]
1238 ) -> Tuple[CacheableOptions, Any]:
1239 return (cls, ())
1241 @hybridmethod
1242 def _generate_cache_key(self) -> Optional[CacheKey]:
1243 return HasCacheKey._generate_cache_key(self)
1246class ExecutableOption(HasCopyInternals):
1247 __slots__ = ()
1249 _annotations: _ImmutableExecuteOptions = util.EMPTY_DICT
1251 __visit_name__: str = "executable_option"
1253 _is_has_cache_key: bool = False
1255 _is_core: bool = True
1257 def _clone(self, **kw):
1258 """Create a shallow copy of this ExecutableOption."""
1259 c = self.__class__.__new__(self.__class__)
1260 c.__dict__ = dict(self.__dict__) # type: ignore[misc]
1261 return c
1264_L = TypeVar("_L", bound=str)
1267class HasSyntaxExtensions(Generic[_L]):
1269 _position_map: Mapping[_L, str]
1271 @_generative
1272 def ext(self, extension: SyntaxExtension) -> Self:
1273 """Applies a SQL syntax extension to this statement.
1275 SQL syntax extensions are :class:`.ClauseElement` objects that define
1276 some vendor-specific syntactical construct that take place in specific
1277 parts of a SQL statement. Examples include vendor extensions like
1278 PostgreSQL / SQLite's "ON DUPLICATE KEY UPDATE", PostgreSQL's
1279 "DISTINCT ON", and MySQL's "LIMIT" that can be applied to UPDATE
1280 and DELETE statements.
1282 .. seealso::
1284 :ref:`examples_syntax_extensions`
1286 :func:`_mysql.limit` - DML LIMIT for MySQL
1288 :func:`_postgresql.distinct_on` - DISTINCT ON for PostgreSQL
1290 .. versionadded:: 2.1
1292 """
1293 extension = coercions.expect(
1294 roles.SyntaxExtensionRole, extension, apply_propagate_attrs=self
1295 )
1296 self._apply_syntax_extension_to_self(extension)
1297 return self
1299 @util.preload_module("sqlalchemy.sql.elements")
1300 def apply_syntax_extension_point(
1301 self,
1302 apply_fn: Callable[[Sequence[ClauseElement]], Sequence[ClauseElement]],
1303 position: _L,
1304 ) -> None:
1305 """Apply a :class:`.SyntaxExtension` to a known extension point.
1307 Should be used only internally by :class:`.SyntaxExtension`.
1309 E.g.::
1311 class Qualify(SyntaxExtension, ClauseElement):
1313 # ...
1315 def apply_to_select(self, select_stmt: Select) -> None:
1316 # append self to existing
1317 select_stmt.apply_extension_point(
1318 lambda existing: [*existing, self], "post_criteria"
1319 )
1322 class ReplaceExt(SyntaxExtension, ClauseElement):
1324 # ...
1326 def apply_to_select(self, select_stmt: Select) -> None:
1327 # replace any existing elements regardless of type
1328 select_stmt.apply_extension_point(
1329 lambda existing: [self], "post_criteria"
1330 )
1333 class ReplaceOfTypeExt(SyntaxExtension, ClauseElement):
1335 # ...
1337 def apply_to_select(self, select_stmt: Select) -> None:
1338 # replace any existing elements of the same type
1339 select_stmt.apply_extension_point(
1340 self.append_replacing_same_type, "post_criteria"
1341 )
1343 :param apply_fn: callable function that will receive a sequence of
1344 :class:`.ClauseElement` that is already populating the extension
1345 point (the sequence is empty if there isn't one), and should return
1346 a new sequence of :class:`.ClauseElement` that will newly populate
1347 that point. The function typically can choose to concatenate the
1348 existing values with the new one, or to replace the values that are
1349 there with a new one by returning a list of a single element, or
1350 to perform more complex operations like removing only the same
1351 type element from the input list of merging already existing elements
1352 of the same type. Some examples are shown in the examples above
1353 :param position: string name of the position to apply to. This
1354 varies per statement type. IDEs should show the possible values
1355 for each statement type as it's typed with a ``typing.Literal`` per
1356 statement.
1358 .. seealso::
1360 :ref:`examples_syntax_extensions`
1362 :meth:`.ext`
1365 """ # noqa: E501
1367 try:
1368 attrname = self._position_map[position]
1369 except KeyError as ke:
1370 raise ValueError(
1371 f"Unknown position {position!r} for {self.__class__} "
1372 f"construct; known positions: "
1373 f"{', '.join(repr(k) for k in self._position_map)}"
1374 ) from ke
1375 else:
1376 ElementList = util.preloaded.sql_elements.ElementList
1377 existing: Optional[ClauseElement] = getattr(self, attrname, None)
1378 if existing is None:
1379 input_seq: Tuple[ClauseElement, ...] = ()
1380 elif isinstance(existing, ElementList):
1381 input_seq = existing.clauses
1382 else:
1383 input_seq = (existing,)
1385 new_seq = apply_fn(input_seq)
1386 assert new_seq, "cannot return empty sequence"
1387 new = new_seq[0] if len(new_seq) == 1 else ElementList(new_seq)
1388 setattr(self, attrname, new)
1390 def _apply_syntax_extension_to_self(
1391 self, extension: SyntaxExtension
1392 ) -> None:
1393 raise NotImplementedError()
1395 def _get_syntax_extensions_as_dict(self) -> Mapping[_L, SyntaxExtension]:
1396 res: Dict[_L, SyntaxExtension] = {}
1397 for name, attr in self._position_map.items():
1398 value = getattr(self, attr)
1399 if value is not None:
1400 res[name] = value
1401 return res
1403 def _set_syntax_extensions(self, **extensions: SyntaxExtension) -> None:
1404 for name, value in extensions.items():
1405 setattr(self, self._position_map[name], value) # type: ignore[index] # noqa: E501
1408class SyntaxExtension(roles.SyntaxExtensionRole):
1409 """Defines a unit that when also extending from :class:`.ClauseElement`
1410 can be applied to SQLAlchemy statements :class:`.Select`,
1411 :class:`_sql.Insert`, :class:`.Update` and :class:`.Delete` making use of
1412 pre-established SQL insertion points within these constructs.
1414 .. versionadded:: 2.1
1416 .. seealso::
1418 :ref:`examples_syntax_extensions`
1420 """
1422 def append_replacing_same_type(
1423 self, existing: Sequence[ClauseElement]
1424 ) -> Sequence[ClauseElement]:
1425 """Utility function that can be used as
1426 :paramref:`_sql.Select.apply_syntax_extension_point.apply_fn`
1427 to remove any other element of the same type in existing and appending
1428 ``self`` to the list.
1430 This is equivalent to::
1432 stmt.apply_syntax_extension_point(
1433 lambda existing: [
1434 *(e for e in existing if not isinstance(e, ReplaceOfTypeExt)),
1435 self,
1436 ],
1437 "post_criteria",
1438 )
1440 .. seealso::
1442 :ref:`examples_syntax_extensions`
1444 :meth:`_sql.Select.apply_syntax_extension_point` and equivalents
1445 in :class:`_dml.Insert`, :class:`_dml.Delete`, :class:`_dml.Update`
1447 """ # noqa: E501
1448 cls = type(self)
1449 return [*(e for e in existing if not isinstance(e, cls)), self] # type: ignore[list-item] # noqa: E501
1451 def apply_to_select(self, select_stmt: Select[Unpack[_Ts]]) -> None:
1452 """Apply this :class:`.SyntaxExtension` to a :class:`.Select`"""
1453 raise NotImplementedError(
1454 f"Extension {type(self).__name__} cannot be applied to select"
1455 )
1457 def apply_to_update(self, update_stmt: Update) -> None:
1458 """Apply this :class:`.SyntaxExtension` to an :class:`.Update`"""
1459 raise NotImplementedError(
1460 f"Extension {type(self).__name__} cannot be applied to update"
1461 )
1463 def apply_to_delete(self, delete_stmt: Delete) -> None:
1464 """Apply this :class:`.SyntaxExtension` to a :class:`.Delete`"""
1465 raise NotImplementedError(
1466 f"Extension {type(self).__name__} cannot be applied to delete"
1467 )
1469 def apply_to_insert(self, insert_stmt: Insert) -> None:
1470 """Apply this :class:`.SyntaxExtension` to an :class:`_sql.Insert`"""
1471 raise NotImplementedError(
1472 f"Extension {type(self).__name__} cannot be applied to insert"
1473 )
1476class Executable(roles.StatementRole):
1477 """Mark a :class:`_expression.ClauseElement` as supporting execution.
1479 :class:`.Executable` is a superclass for all "statement" types
1480 of objects, including :func:`select`, :func:`delete`, :func:`update`,
1481 :func:`insert`, :func:`text`.
1483 """
1485 supports_execution: bool = True
1486 _execution_options: _ImmutableExecuteOptions = util.EMPTY_DICT
1487 _is_default_generator: bool = False
1488 _with_options: Tuple[ExecutableOption, ...] = ()
1489 _compile_state_funcs: Tuple[
1490 Tuple[Callable[[CompileState], None], Any], ...
1491 ] = ()
1492 _compile_options: Optional[Union[Type[CacheableOptions], CacheableOptions]]
1494 _executable_traverse_internals = [
1495 ("_with_options", InternalTraversal.dp_executable_options),
1496 (
1497 "_compile_state_funcs",
1498 ExtendedInternalTraversal.dp_compile_state_funcs,
1499 ),
1500 ("_propagate_attrs", ExtendedInternalTraversal.dp_propagate_attrs),
1501 ]
1503 is_select: bool = False
1504 is_from_statement: bool = False
1505 is_update: bool = False
1506 is_insert: bool = False
1507 is_text: bool = False
1508 is_delete: bool = False
1509 is_dml: bool = False
1511 if TYPE_CHECKING:
1512 __visit_name__: str
1514 def _compile_w_cache(
1515 self,
1516 dialect: Dialect,
1517 *,
1518 compiled_cache: Optional[CompiledCacheType],
1519 column_keys: List[str],
1520 for_executemany: bool = False,
1521 schema_translate_map: Optional[SchemaTranslateMapType] = None,
1522 **kw: Any,
1523 ) -> tuple[
1524 Compiled,
1525 Sequence[BindParameter[Any]] | None,
1526 _CoreSingleExecuteParams | None,
1527 CacheStats,
1528 ]: ...
1530 def _execute_on_connection(
1531 self,
1532 connection: Connection,
1533 distilled_params: _CoreMultiExecuteParams,
1534 execution_options: CoreExecuteOptionsParameter,
1535 ) -> CursorResult[Any]: ...
1537 def _execute_on_scalar(
1538 self,
1539 connection: Connection,
1540 distilled_params: _CoreMultiExecuteParams,
1541 execution_options: CoreExecuteOptionsParameter,
1542 ) -> Any: ...
1544 @util.ro_non_memoized_property
1545 def _all_selected_columns(self) -> _SelectIterable:
1546 raise NotImplementedError()
1548 @property
1549 def _effective_plugin_target(self) -> str:
1550 return self.__visit_name__
1552 @_generative
1553 def options(self, *options: ExecutableOption) -> Self:
1554 """Apply options to this statement.
1556 In the general sense, options are any kind of Python object
1557 that can be interpreted by systems that consume the statement outside
1558 of the regular SQL compiler chain. Specifically, these options are
1559 the ORM level options that apply "eager load" and other loading
1560 behaviors to an ORM query.
1562 For background on specific kinds of options for specific kinds of
1563 statements, refer to the documentation for those option objects.
1565 .. versionchanged:: 1.4 - added :meth:`.Executable.options` to
1566 Core statement objects towards the goal of allowing unified
1567 Core / ORM querying capabilities.
1569 .. seealso::
1571 :ref:`loading_columns` - refers to options specific to the usage
1572 of ORM queries
1574 :ref:`relationship_loader_options` - refers to options specific
1575 to the usage of ORM queries
1577 """
1578 self._with_options += tuple(
1579 coercions.expect(roles.ExecutableOptionRole, opt)
1580 for opt in options
1581 )
1582 return self
1584 @_generative
1585 def _set_compile_options(self, compile_options: CacheableOptions) -> Self:
1586 """Assign the compile options to a new value.
1588 :param compile_options: appropriate CacheableOptions structure
1590 """
1592 self._compile_options = compile_options
1593 return self
1595 @_generative
1596 def _update_compile_options(self, options: CacheableOptions) -> Self:
1597 """update the _compile_options with new keys."""
1599 assert self._compile_options is not None
1600 self._compile_options += options
1601 return self
1603 @_generative
1604 def _add_compile_state_func(
1605 self,
1606 callable_: Callable[[CompileState], None],
1607 cache_args: Any,
1608 ) -> Self:
1609 """Add a compile state function to this statement.
1611 When using the ORM only, these are callable functions that will
1612 be given the CompileState object upon compilation.
1614 A second argument cache_args is required, which will be combined with
1615 the ``__code__`` identity of the function itself in order to produce a
1616 cache key.
1618 """
1619 self._compile_state_funcs += ((callable_, cache_args),)
1620 return self
1622 @overload
1623 def execution_options(
1624 self,
1625 *,
1626 compiled_cache: Optional[CompiledCacheType] = ...,
1627 logging_token: str = ...,
1628 isolation_level: IsolationLevel = ...,
1629 no_parameters: bool = False,
1630 stream_results: bool = False,
1631 max_row_buffer: int = ...,
1632 yield_per: int = ...,
1633 driver_column_names: bool = ...,
1634 insertmanyvalues_page_size: int = ...,
1635 schema_translate_map: Optional[SchemaTranslateMapType] = ...,
1636 populate_existing: bool = False,
1637 autoflush: bool = False,
1638 synchronize_session: SynchronizeSessionArgument = ...,
1639 dml_strategy: DMLStrategyArgument = ...,
1640 render_nulls: bool = ...,
1641 is_delete_using: bool = ...,
1642 is_update_from: bool = ...,
1643 preserve_rowcount: bool = False,
1644 **opt: Any,
1645 ) -> Self: ...
1647 @overload
1648 def execution_options(self, **opt: Any) -> Self: ...
1650 @_generative
1651 def execution_options(self, **kw: Any) -> Self:
1652 """Set non-SQL options for the statement which take effect during
1653 execution.
1655 Execution options can be set at many scopes, including per-statement,
1656 per-connection, or per execution, using methods such as
1657 :meth:`_engine.Connection.execution_options` and parameters which
1658 accept a dictionary of options such as
1659 :paramref:`_engine.Connection.execute.execution_options` and
1660 :paramref:`_orm.Session.execute.execution_options`.
1662 The primary characteristic of an execution option, as opposed to
1663 other kinds of options such as ORM loader options, is that
1664 **execution options never affect the compiled SQL of a query, only
1665 things that affect how the SQL statement itself is invoked or how
1666 results are fetched**. That is, execution options are not part of
1667 what's accommodated by SQL compilation nor are they considered part of
1668 the cached state of a statement.
1670 The :meth:`_sql.Executable.execution_options` method is
1671 :term:`generative`, as
1672 is the case for the method as applied to the :class:`_engine.Engine`
1673 and :class:`_orm.Query` objects, which means when the method is called,
1674 a copy of the object is returned, which applies the given parameters to
1675 that new copy, but leaves the original unchanged::
1677 statement = select(table.c.x, table.c.y)
1678 new_statement = statement.execution_options(my_option=True)
1680 An exception to this behavior is the :class:`_engine.Connection`
1681 object, where the :meth:`_engine.Connection.execution_options` method
1682 is explicitly **not** generative.
1684 The kinds of options that may be passed to
1685 :meth:`_sql.Executable.execution_options` and other related methods and
1686 parameter dictionaries include parameters that are explicitly consumed
1687 by SQLAlchemy Core or ORM, as well as arbitrary keyword arguments not
1688 defined by SQLAlchemy, which means the methods and/or parameter
1689 dictionaries may be used for user-defined parameters that interact with
1690 custom code, which may access the parameters using methods such as
1691 :meth:`_sql.Executable.get_execution_options` and
1692 :meth:`_engine.Connection.get_execution_options`, or within selected
1693 event hooks using a dedicated ``execution_options`` event parameter
1694 such as
1695 :paramref:`_events.ConnectionEvents.before_execute.execution_options`
1696 or :attr:`_orm.ORMExecuteState.execution_options`, e.g.::
1698 from sqlalchemy import event
1701 @event.listens_for(some_engine, "before_execute")
1702 def _process_opt(conn, statement, multiparams, params, execution_options):
1703 "run a SQL function before invoking a statement"
1705 if execution_options.get("do_special_thing", False):
1706 conn.exec_driver_sql("run_special_function()")
1708 Within the scope of options that are explicitly recognized by
1709 SQLAlchemy, most apply to specific classes of objects and not others.
1710 The most common execution options include:
1712 * :paramref:`_engine.Connection.execution_options.isolation_level` -
1713 sets the isolation level for a connection or a class of connections
1714 via an :class:`_engine.Engine`. This option is accepted only
1715 by :class:`_engine.Connection` or :class:`_engine.Engine`.
1717 * :paramref:`_engine.Connection.execution_options.stream_results` -
1718 indicates results should be fetched using a server side cursor;
1719 this option is accepted by :class:`_engine.Connection`, by the
1720 :paramref:`_engine.Connection.execute.execution_options` parameter
1721 on :meth:`_engine.Connection.execute`, and additionally by
1722 :meth:`_sql.Executable.execution_options` on a SQL statement object,
1723 as well as by ORM constructs like :meth:`_orm.Session.execute`.
1725 * :paramref:`_engine.Connection.execution_options.compiled_cache` -
1726 indicates a dictionary that will serve as the
1727 :ref:`SQL compilation cache <sql_caching>`
1728 for a :class:`_engine.Connection` or :class:`_engine.Engine`, as
1729 well as for ORM methods like :meth:`_orm.Session.execute`.
1730 Can be passed as ``None`` to disable caching for statements.
1731 This option is not accepted by
1732 :meth:`_sql.Executable.execution_options` as it is inadvisable to
1733 carry along a compilation cache within a statement object.
1735 * :paramref:`_engine.Connection.execution_options.schema_translate_map`
1736 - a mapping of schema names used by the
1737 :ref:`Schema Translate Map <schema_translating>` feature, accepted
1738 by :class:`_engine.Connection`, :class:`_engine.Engine`,
1739 :class:`_sql.Executable`, as well as by ORM constructs
1740 like :meth:`_orm.Session.execute`.
1742 .. seealso::
1744 :meth:`_engine.Connection.execution_options`
1746 :paramref:`_engine.Connection.execute.execution_options`
1748 :paramref:`_orm.Session.execute.execution_options`
1750 :ref:`orm_queryguide_execution_options` - documentation on all
1751 ORM-specific execution options
1753 """ # noqa: E501
1754 if "isolation_level" in kw:
1755 raise exc.ArgumentError(
1756 "'isolation_level' execution option may only be specified "
1757 "on Connection.execution_options(), or "
1758 "per-engine using the isolation_level "
1759 "argument to create_engine()."
1760 )
1761 if "compiled_cache" in kw:
1762 raise exc.ArgumentError(
1763 "'compiled_cache' execution option may only be specified "
1764 "on Connection.execution_options(), not per statement."
1765 )
1766 self._execution_options = self._execution_options.union(kw)
1767 return self
1769 def get_execution_options(self) -> _ExecuteOptions:
1770 """Get the non-SQL options which will take effect during execution.
1772 .. seealso::
1774 :meth:`.Executable.execution_options`
1775 """
1776 return self._execution_options
1779class ExecutableStatement(Executable):
1780 """Executable subclass that implements a lightweight version of ``params``
1781 that avoids a full cloned traverse.
1783 .. versionadded:: 2.1
1785 """
1787 _params: util.immutabledict[str, Any] = EMPTY_DICT
1789 _executable_traverse_internals = (
1790 Executable._executable_traverse_internals
1791 + [("_params", InternalTraversal.dp_params)]
1792 )
1794 @_generative
1795 def params(
1796 self,
1797 __optionaldict: _CoreSingleExecuteParams | None = None,
1798 /,
1799 **kwargs: Any,
1800 ) -> Self:
1801 """Return a copy with the provided bindparam values.
1803 Returns a copy of this Executable with bindparam values set
1804 to the given dictionary::
1806 >>> clause = column("x") + bindparam("foo")
1807 >>> print(clause.compile().params)
1808 {'foo': None}
1809 >>> print(clause.params({"foo": 7}).compile().params)
1810 {'foo': 7}
1812 """
1813 if __optionaldict:
1814 kwargs.update(__optionaldict)
1815 self._params = (
1816 util.immutabledict(kwargs)
1817 if not self._params
1818 else self._params | kwargs
1819 )
1820 return self
1823class SchemaEventTarget(event.EventTarget):
1824 """Base class for elements that are the targets of :class:`.DDLEvents`
1825 events.
1827 This includes :class:`.SchemaItem` as well as :class:`.SchemaType`.
1829 """
1831 dispatch: dispatcher[SchemaEventTarget]
1833 def _set_parent(self, parent: SchemaEventTarget, **kw: Any) -> None:
1834 """Associate with this SchemaEvent's parent object."""
1836 def _set_parent_with_dispatch(
1837 self, parent: SchemaEventTarget, **kw: Any
1838 ) -> None:
1839 self.dispatch.before_parent_attach(self, parent)
1840 self._set_parent(parent, **kw)
1841 self.dispatch.after_parent_attach(self, parent)
1844class SchemaVisitable(SchemaEventTarget, visitors.Visitable):
1845 """Base class for elements that are targets of a :class:`.SchemaVisitor`.
1847 .. versionadded:: 2.0.41
1849 """
1852class SchemaVisitor(ClauseVisitor):
1853 """Define the visiting for ``SchemaItem`` and more
1854 generally ``SchemaVisitable`` objects.
1856 """
1858 __traverse_options__: Dict[str, Any] = {"schema_visitor": True}
1861class _SentinelDefaultCharacterization(Enum):
1862 NONE = "none"
1863 UNKNOWN = "unknown"
1864 CLIENTSIDE = "clientside"
1865 SENTINEL_DEFAULT = "sentinel_default"
1866 SERVERSIDE = "serverside"
1867 IDENTITY = "identity"
1868 SEQUENCE = "sequence"
1869 MONOTONIC_FUNCTION = "monotonic"
1872class _SentinelColumnCharacterization(NamedTuple):
1873 columns: Optional[Sequence[Column[Any]]] = None
1874 is_explicit: bool = False
1875 is_autoinc: bool = False
1876 default_characterization: _SentinelDefaultCharacterization = (
1877 _SentinelDefaultCharacterization.NONE
1878 )
1881_COLKEY = TypeVar("_COLKEY", Union[None, str], str)
1883_COL_co = TypeVar("_COL_co", bound="ColumnElement[Any]", covariant=True)
1884_COL = TypeVar("_COL", bound="ColumnElement[Any]")
1887class _ColumnMetrics(Generic[_COL_co]):
1888 __slots__ = ("column",)
1890 column: _COL_co
1892 def __init__(
1893 self, collection: ColumnCollection[Any, _COL_co], col: _COL_co
1894 ) -> None:
1895 self.column = col
1897 # proxy_index being non-empty means it was initialized.
1898 # so we need to update it
1899 pi = collection._proxy_index
1900 if pi:
1901 for eps_col in col._expanded_proxy_set:
1902 pi[eps_col].add(self)
1904 def get_expanded_proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1905 return self.column._expanded_proxy_set
1907 def dispose(self, collection: ColumnCollection[_COLKEY, _COL_co]) -> None:
1908 pi = collection._proxy_index
1909 if not pi:
1910 return
1911 for col in self.column._expanded_proxy_set:
1912 colset = pi.get(col, None)
1913 if colset:
1914 colset.discard(self)
1915 if colset is not None and not colset:
1916 del pi[col]
1918 def embedded(
1919 self,
1920 target_set: Union[
1921 Set[ColumnElement[Any]], FrozenSet[ColumnElement[Any]]
1922 ],
1923 ) -> bool:
1924 expanded_proxy_set = self.column._expanded_proxy_set
1925 for t in target_set.difference(expanded_proxy_set):
1926 if not expanded_proxy_set.intersection(_expand_cloned([t])):
1927 return False
1928 return True
1931class ColumnCollection(Generic[_COLKEY, _COL_co]):
1932 """Base class for collection of :class:`_expression.ColumnElement`
1933 instances, typically for :class:`_sql.FromClause` objects.
1935 The :class:`_sql.ColumnCollection` object is most commonly available
1936 as the :attr:`_schema.Table.c` or :attr:`_schema.Table.columns` collection
1937 on the :class:`_schema.Table` object, introduced at
1938 :ref:`metadata_tables_and_columns`.
1940 The :class:`_expression.ColumnCollection` has both mapping- and sequence-
1941 like behaviors. A :class:`_expression.ColumnCollection` usually stores
1942 :class:`_schema.Column` objects, which are then accessible both via mapping
1943 style access as well as attribute access style.
1945 To access :class:`_schema.Column` objects using ordinary attribute-style
1946 access, specify the name like any other object attribute, such as below
1947 a column named ``employee_name`` is accessed::
1949 >>> employee_table.c.employee_name
1951 To access columns that have names with special characters or spaces,
1952 index-style access is used, such as below which illustrates a column named
1953 ``employee ' payment`` is accessed::
1955 >>> employee_table.c["employee ' payment"]
1957 As the :class:`_sql.ColumnCollection` object provides a Python dictionary
1958 interface, common dictionary method names like
1959 :meth:`_sql.ColumnCollection.keys`, :meth:`_sql.ColumnCollection.values`,
1960 and :meth:`_sql.ColumnCollection.items` are available, which means that
1961 database columns that are keyed under these names also need to use indexed
1962 access::
1964 >>> employee_table.c["values"]
1967 The name for which a :class:`_schema.Column` would be present is normally
1968 that of the :paramref:`_schema.Column.key` parameter. In some contexts,
1969 such as a :class:`_sql.Select` object that uses a label style set
1970 using the :meth:`_sql.Select.set_label_style` method, a column of a certain
1971 key may instead be represented under a particular label name such
1972 as ``tablename_columnname``::
1974 >>> from sqlalchemy import select, column, table
1975 >>> from sqlalchemy import LABEL_STYLE_TABLENAME_PLUS_COL
1976 >>> t = table("t", column("c"))
1977 >>> stmt = select(t).set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL)
1978 >>> subq = stmt.subquery()
1979 >>> subq.c.t_c
1980 <sqlalchemy.sql.elements.ColumnClause at 0x7f59dcf04fa0; t_c>
1982 :class:`.ColumnCollection` also indexes the columns in order and allows
1983 them to be accessible by their integer position::
1985 >>> cc[0]
1986 Column('x', Integer(), table=None)
1987 >>> cc[1]
1988 Column('y', Integer(), table=None)
1990 .. versionadded:: 1.4 :class:`_expression.ColumnCollection`
1991 allows integer-based
1992 index access to the collection.
1994 Iterating the collection yields the column expressions in order::
1996 >>> list(cc)
1997 [Column('x', Integer(), table=None),
1998 Column('y', Integer(), table=None)]
2000 The :class:`_expression.ColumnCollection` base class is read-only.
2001 For mutation operations, the :class:`.WriteableColumnCollection` subclass
2002 provides methods such as :meth:`.WriteableColumnCollection.add`.
2003 A special subclass :class:`.DedupeColumnCollection` exists which
2004 maintains SQLAlchemy's older behavior of not allowing duplicates; this
2005 collection is used for schema level objects like :class:`_schema.Table`
2006 and :class:`.PrimaryKeyConstraint` where this deduping is helpful.
2007 The :class:`.DedupeColumnCollection` class also has additional mutation
2008 methods as the schema constructs have more use cases that require removal
2009 and replacement of columns.
2011 .. versionchanged:: 1.4 :class:`_expression.ColumnCollection`
2012 now stores duplicate
2013 column keys as well as the same column in multiple positions. The
2014 :class:`.DedupeColumnCollection` class is added to maintain the
2015 former behavior in those cases where deduplication as well as
2016 additional replace/remove operations are needed.
2018 .. versionchanged:: 2.1 :class:`_expression.ColumnCollection` is now
2019 a read-only base class. Mutation operations are available through
2020 :class:`.WriteableColumnCollection` and :class:`.DedupeColumnCollection`
2021 subclasses.
2024 """
2026 __slots__ = ("_collection", "_index", "_colset", "_proxy_index")
2028 _collection: List[Tuple[_COLKEY, _COL_co, _ColumnMetrics[_COL_co]]]
2029 _index: Dict[Union[None, str, int], Tuple[_COLKEY, _COL_co]]
2030 _colset: Set[_COL_co]
2031 _proxy_index: Dict[ColumnElement[Any], Set[_ColumnMetrics[_COL_co]]]
2033 def __init__(self) -> None:
2034 raise TypeError(
2035 "ColumnCollection is an abstract base class and cannot be "
2036 "instantiated directly. Use WriteableColumnCollection or "
2037 "DedupeColumnCollection instead."
2038 )
2040 @util.preload_module("sqlalchemy.sql.elements")
2041 def __clause_element__(self) -> ClauseList:
2042 elements = util.preloaded.sql_elements
2044 return elements.ClauseList(
2045 _literal_as_text_role=roles.ColumnsClauseRole,
2046 group=False,
2047 *self._all_columns,
2048 )
2050 @property
2051 def _all_columns(self) -> List[_COL_co]:
2052 return [col for (_, col, _) in self._collection]
2054 def keys(self) -> List[_COLKEY]:
2055 """Return a sequence of string key names for all columns in this
2056 collection."""
2057 return [k for (k, _, _) in self._collection]
2059 def values(self) -> List[_COL_co]:
2060 """Return a sequence of :class:`_sql.ColumnClause` or
2061 :class:`_schema.Column` objects for all columns in this
2062 collection."""
2063 return [col for (_, col, _) in self._collection]
2065 def items(self) -> List[Tuple[_COLKEY, _COL_co]]:
2066 """Return a sequence of (key, column) tuples for all columns in this
2067 collection each consisting of a string key name and a
2068 :class:`_sql.ColumnClause` or
2069 :class:`_schema.Column` object.
2070 """
2072 return [(k, col) for (k, col, _) in self._collection]
2074 def __bool__(self) -> bool:
2075 return bool(self._collection)
2077 def __len__(self) -> int:
2078 return len(self._collection)
2080 def __iter__(self) -> Iterator[_COL_co]:
2081 # turn to a list first to maintain over a course of changes
2082 return iter([col for _, col, _ in self._collection])
2084 @overload
2085 def __getitem__(self, key: Union[str, int]) -> _COL_co: ...
2087 @overload
2088 def __getitem__(
2089 self, key: Union[Tuple[Union[str, int], ...], slice]
2090 ) -> ReadOnlyColumnCollection[_COLKEY, _COL_co]: ...
2092 def __getitem__(
2093 self, key: Union[str, int, slice, Tuple[Union[str, int], ...]]
2094 ) -> Union[ReadOnlyColumnCollection[_COLKEY, _COL_co], _COL_co]:
2095 try:
2096 if isinstance(key, (tuple, slice)):
2097 if isinstance(key, slice):
2098 cols = (
2099 (sub_key, col)
2100 for (sub_key, col, _) in self._collection[key]
2101 )
2102 else:
2103 cols = (self._index[sub_key] for sub_key in key)
2105 return WriteableColumnCollection(cols).as_readonly()
2106 else:
2107 return self._index[key][1]
2108 except KeyError as err:
2109 if isinstance(err.args[0], int):
2110 raise IndexError(err.args[0]) from err
2111 else:
2112 raise
2114 def __getattr__(self, key: str) -> _COL_co:
2115 try:
2116 return self._index[key][1]
2117 except KeyError as err:
2118 raise AttributeError(key) from err
2120 def __contains__(self, key: str) -> bool:
2121 if key not in self._index:
2122 if not isinstance(key, str):
2123 raise exc.ArgumentError(
2124 "__contains__ requires a string argument"
2125 )
2126 return False
2127 else:
2128 return True
2130 def compare(self, other: ColumnCollection[_COLKEY, _COL_co]) -> bool:
2131 """Compare this :class:`_expression.ColumnCollection` to another
2132 based on the names of the keys"""
2134 for l, r in zip_longest(self, other):
2135 if l is not r:
2136 return False
2137 else:
2138 return True
2140 def __eq__(self, other: Any) -> bool:
2141 return self.compare(other)
2143 @overload
2144 def get(self, key: str, default: None = None) -> Optional[_COL_co]: ...
2146 @overload
2147 def get(self, key: str, default: _COL) -> Union[_COL_co, _COL]: ...
2149 def get(
2150 self, key: str, default: Optional[_COL] = None
2151 ) -> Optional[Union[_COL_co, _COL]]:
2152 """Get a :class:`_sql.ColumnClause` or :class:`_schema.Column` object
2153 based on a string key name from this
2154 :class:`_expression.ColumnCollection`."""
2156 if key in self._index:
2157 return self._index[key][1]
2158 else:
2159 return default
2161 def __str__(self) -> str:
2162 return "%s(%s)" % (
2163 self.__class__.__name__,
2164 ", ".join(str(c) for c in self),
2165 )
2167 # https://github.com/python/mypy/issues/4266
2168 __hash__: Optional[int] = None # type: ignore[assignment]
2170 def contains_column(self, col: ColumnElement[Any]) -> bool:
2171 """Checks if a column object exists in this collection"""
2172 if col not in self._colset:
2173 if isinstance(col, str):
2174 raise exc.ArgumentError(
2175 "contains_column cannot be used with string arguments. "
2176 "Use ``col_name in table.c`` instead."
2177 )
2178 return False
2179 else:
2180 return True
2182 def _as_readonly(self) -> ReadOnlyColumnCollection[_COLKEY, _COL_co]:
2183 raise NotImplementedError()
2185 def corresponding_column(
2186 self, column: _COL, require_embedded: bool = False
2187 ) -> Optional[Union[_COL, _COL_co]]:
2188 """Given a :class:`_expression.ColumnElement`, return the exported
2189 :class:`_expression.ColumnElement` object from this
2190 :class:`_expression.ColumnCollection`
2191 which corresponds to that original :class:`_expression.ColumnElement`
2192 via a common
2193 ancestor column.
2195 :param column: the target :class:`_expression.ColumnElement`
2196 to be matched.
2198 :param require_embedded: only return corresponding columns for
2199 the given :class:`_expression.ColumnElement`, if the given
2200 :class:`_expression.ColumnElement`
2201 is actually present within a sub-element
2202 of this :class:`_expression.Selectable`.
2203 Normally the column will match if
2204 it merely shares a common ancestor with one of the exported
2205 columns of this :class:`_expression.Selectable`.
2207 .. seealso::
2209 :meth:`_expression.Selectable.corresponding_column`
2210 - invokes this method
2211 against the collection returned by
2212 :attr:`_expression.Selectable.exported_columns`.
2214 .. versionchanged:: 1.4 the implementation for ``corresponding_column``
2215 was moved onto the :class:`_expression.ColumnCollection` itself.
2217 """
2218 raise NotImplementedError()
2221class WriteableColumnCollection(ColumnCollection[_COLKEY, _COL_co]):
2222 """A :class:`_sql.ColumnCollection` that allows mutation operations.
2224 This is the writable form of :class:`_sql.ColumnCollection` that
2225 implements methods such as :meth:`.add`, :meth:`.remove`, :meth:`.update`,
2226 and :meth:`.clear`.
2228 This class is used internally for building column collections during
2229 construction of SQL constructs. For schema-level objects that require
2230 deduplication behavior, use :class:`.DedupeColumnCollection`.
2232 .. versionadded:: 2.1
2234 """
2236 __slots__ = ()
2238 def __init__(
2239 self, columns: Optional[Iterable[Tuple[_COLKEY, _COL_co]]] = None
2240 ):
2241 object.__setattr__(self, "_colset", set())
2242 object.__setattr__(self, "_index", {})
2243 object.__setattr__(
2244 self, "_proxy_index", collections.defaultdict(util.OrderedSet)
2245 )
2246 object.__setattr__(self, "_collection", [])
2247 if columns:
2248 self._initial_populate(columns)
2250 def _initial_populate(
2251 self, iter_: Iterable[Tuple[_COLKEY, _COL_co]]
2252 ) -> None:
2253 self._populate_separate_keys(iter_)
2255 def _populate_separate_keys(
2256 self, iter_: Iterable[Tuple[_COLKEY, _COL_co]]
2257 ) -> None:
2258 """populate from an iterator of (key, column)"""
2260 self._collection[:] = collection = [
2261 (k, c, _ColumnMetrics(self, c)) for k, c in iter_
2262 ]
2263 self._colset.update(c._deannotate() for _, c, _ in collection)
2264 self._index.update(
2265 {idx: (k, c) for idx, (k, c, _) in enumerate(collection)}
2266 )
2267 self._index.update({k: (k, col) for k, col, _ in reversed(collection)})
2269 def __getstate__(self) -> Dict[str, Any]:
2270 return {
2271 "_collection": [(k, c) for k, c, _ in self._collection],
2272 "_index": self._index,
2273 }
2275 def __setstate__(self, state: Dict[str, Any]) -> None:
2276 object.__setattr__(self, "_index", state["_index"])
2277 object.__setattr__(
2278 self, "_proxy_index", collections.defaultdict(util.OrderedSet)
2279 )
2280 object.__setattr__(
2281 self,
2282 "_collection",
2283 [
2284 (k, c, _ColumnMetrics(self, c))
2285 for (k, c) in state["_collection"]
2286 ],
2287 )
2288 object.__setattr__(
2289 self, "_colset", {col for k, col, _ in self._collection}
2290 )
2292 def add(
2293 self,
2294 column: ColumnElement[Any],
2295 key: Optional[_COLKEY] = None,
2296 ) -> None:
2297 """Add a column to this :class:`_sql.WriteableColumnCollection`.
2299 .. note::
2301 This method is **not normally used by user-facing code**, as the
2302 :class:`_sql.WriteableColumnCollection` is usually part of an
2303 existing object such as a :class:`_schema.Table`. To add a
2304 :class:`_schema.Column` to an existing :class:`_schema.Table`
2305 object, use the :meth:`_schema.Table.append_column` method.
2307 """
2308 colkey: _COLKEY
2310 if key is None:
2311 colkey = column.key # type: ignore[assignment]
2312 else:
2313 colkey = key
2315 l = len(self._collection)
2317 # don't really know how this part is supposed to work w/ the
2318 # covariant thing
2320 _column = cast(_COL_co, column)
2322 self._collection.append(
2323 (colkey, _column, _ColumnMetrics(self, _column))
2324 )
2325 self._colset.add(_column._deannotate())
2327 self._index[l] = (colkey, _column)
2328 if colkey not in self._index:
2329 self._index[colkey] = (colkey, _column)
2331 def _as_readonly(self) -> ReadOnlyColumnCollection[_COLKEY, _COL_co]:
2332 return ReadOnlyColumnCollection(self)
2334 def as_readonly(self) -> ReadOnlyColumnCollection[_COLKEY, _COL_co]:
2335 """Return a "read only" form of this
2336 :class:`_sql.WriteableColumnCollection`."""
2338 return self._as_readonly()
2340 def _init_proxy_index(self) -> None:
2341 """populate the "proxy index", if empty.
2343 proxy index is added in 2.0 to provide more efficient operation
2344 for the corresponding_column() method.
2346 For reasons of both time to construct new .c collections as well as
2347 memory conservation for large numbers of large .c collections, the
2348 proxy_index is only filled if corresponding_column() is called. once
2349 filled it stays that way, and new _ColumnMetrics objects created after
2350 that point will populate it with new data. Note this case would be
2351 unusual, if not nonexistent, as it means a .c collection is being
2352 mutated after corresponding_column() were used, however it is tested in
2353 test/base/test_utils.py.
2355 """
2356 pi = self._proxy_index
2357 if pi:
2358 return
2360 for _, _, metrics in self._collection:
2361 eps = metrics.column._expanded_proxy_set
2363 for eps_col in eps:
2364 pi[eps_col].add(metrics)
2366 def corresponding_column(
2367 self, column: _COL, require_embedded: bool = False
2368 ) -> Optional[Union[_COL, _COL_co]]:
2369 """Given a :class:`_expression.ColumnElement`, return the exported
2370 :class:`_expression.ColumnElement` object from this
2371 :class:`_expression.ColumnCollection`
2372 which corresponds to that original :class:`_expression.ColumnElement`
2373 via a common
2374 ancestor column.
2376 See :meth:`.ColumnCollection.corresponding_column` for parameter
2377 information.
2379 """
2380 # TODO: cython candidate
2382 # don't dig around if the column is locally present
2383 if column in self._colset:
2384 return column
2386 selected_intersection, selected_metrics = None, None
2387 target_set = column.proxy_set
2389 pi = self._proxy_index
2390 if not pi:
2391 self._init_proxy_index()
2393 for current_metrics in (
2394 mm for ts in target_set if ts in pi for mm in pi[ts]
2395 ):
2396 if not require_embedded or current_metrics.embedded(target_set):
2397 if selected_metrics is None:
2398 # no corresponding column yet, pick this one.
2399 selected_metrics = current_metrics
2400 continue
2402 current_intersection = target_set.intersection(
2403 current_metrics.column._expanded_proxy_set
2404 )
2405 if selected_intersection is None:
2406 selected_intersection = target_set.intersection(
2407 selected_metrics.column._expanded_proxy_set
2408 )
2410 if len(current_intersection) > len(selected_intersection):
2411 # 'current' has a larger field of correspondence than
2412 # 'selected'. i.e. selectable.c.a1_x->a1.c.x->table.c.x
2413 # matches a1.c.x->table.c.x better than
2414 # selectable.c.x->table.c.x does.
2416 selected_metrics = current_metrics
2417 selected_intersection = current_intersection
2418 elif current_intersection == selected_intersection:
2419 # they have the same field of correspondence. see
2420 # which proxy_set has fewer columns in it, which
2421 # indicates a closer relationship with the root
2422 # column. Also take into account the "weight"
2423 # attribute which CompoundSelect() uses to give
2424 # higher precedence to columns based on vertical
2425 # position in the compound statement, and discard
2426 # columns that have no reference to the target
2427 # column (also occurs with CompoundSelect)
2429 selected_col_distance = sum(
2430 [
2431 sc._annotations.get("weight", 1)
2432 for sc in (
2433 selected_metrics.column._uncached_proxy_list()
2434 )
2435 if sc.shares_lineage(column)
2436 ],
2437 )
2438 current_col_distance = sum(
2439 [
2440 sc._annotations.get("weight", 1)
2441 for sc in (
2442 current_metrics.column._uncached_proxy_list()
2443 )
2444 if sc.shares_lineage(column)
2445 ],
2446 )
2447 if current_col_distance < selected_col_distance:
2448 selected_metrics = current_metrics
2449 selected_intersection = current_intersection
2451 return selected_metrics.column if selected_metrics else None
2454_NAMEDCOL = TypeVar("_NAMEDCOL", bound="NamedColumn[Any]")
2457class DedupeColumnCollection(WriteableColumnCollection[str, _NAMEDCOL]):
2458 """A :class:`_expression.ColumnCollection`
2459 that maintains deduplicating behavior.
2461 This is useful by schema level objects such as :class:`_schema.Table` and
2462 :class:`.PrimaryKeyConstraint`. The collection includes more
2463 sophisticated mutator methods as well to suit schema objects which
2464 require mutable column collections.
2466 .. versionadded:: 1.4
2468 """
2470 def add( # type: ignore[override]
2471 self,
2472 column: _NAMEDCOL,
2473 key: Optional[str] = None,
2474 *,
2475 index: Optional[int] = None,
2476 ) -> None:
2477 if key is not None and column.key != key:
2478 raise exc.ArgumentError(
2479 "DedupeColumnCollection requires columns be under "
2480 "the same key as their .key"
2481 )
2482 key = column.key
2484 if key is None:
2485 raise exc.ArgumentError(
2486 "Can't add unnamed column to column collection"
2487 )
2489 if key in self._index:
2490 existing = self._index[key][1]
2492 if existing is column:
2493 return
2495 self.replace(column, index=index)
2497 # pop out memoized proxy_set as this
2498 # operation may very well be occurring
2499 # in a _make_proxy operation
2500 util.memoized_property.reset(column, "proxy_set")
2501 else:
2502 self._append_new_column(key, column, index=index)
2504 def _append_new_column(
2505 self, key: str, named_column: _NAMEDCOL, *, index: Optional[int] = None
2506 ) -> None:
2507 collection_length = len(self._collection)
2509 if index is None:
2510 l = collection_length
2511 else:
2512 if index < 0:
2513 index = max(0, collection_length + index)
2514 l = index
2516 if index is None:
2517 self._collection.append(
2518 (key, named_column, _ColumnMetrics(self, named_column))
2519 )
2520 else:
2521 self._collection.insert(
2522 index, (key, named_column, _ColumnMetrics(self, named_column))
2523 )
2525 self._colset.add(named_column._deannotate())
2527 if index is not None:
2528 for idx in reversed(range(index, collection_length)):
2529 self._index[idx + 1] = self._index[idx]
2531 self._index[l] = (key, named_column)
2532 self._index[key] = (key, named_column)
2534 def _populate_separate_keys(
2535 self, iter_: Iterable[Tuple[str, _NAMEDCOL]]
2536 ) -> None:
2537 """populate from an iterator of (key, column)"""
2538 cols = list(iter_)
2540 replace_col = []
2541 for k, col in cols:
2542 if col.key != k:
2543 raise exc.ArgumentError(
2544 "DedupeColumnCollection requires columns be under "
2545 "the same key as their .key"
2546 )
2547 if col.name in self._index and col.key != col.name:
2548 replace_col.append(col)
2549 elif col.key in self._index:
2550 replace_col.append(col)
2551 else:
2552 self._index[k] = (k, col)
2553 self._collection.append((k, col, _ColumnMetrics(self, col)))
2554 self._colset.update(c._deannotate() for (k, c, _) in self._collection)
2556 self._index.update(
2557 (idx, (k, c)) for idx, (k, c, _) in enumerate(self._collection)
2558 )
2559 for col in replace_col:
2560 self.replace(col)
2562 def extend(self, iter_: Iterable[_NAMEDCOL]) -> None:
2563 self._populate_separate_keys((col.key, col) for col in iter_)
2565 def remove(self, column: _NAMEDCOL) -> None:
2566 if column not in self._colset:
2567 raise ValueError(
2568 "Can't remove column %r; column is not in this collection"
2569 % column
2570 )
2571 del self._index[column.key]
2572 self._colset.remove(column)
2573 self._collection[:] = [
2574 (k, c, metrics)
2575 for (k, c, metrics) in self._collection
2576 if c is not column
2577 ]
2578 for metrics in self._proxy_index.get(column, ()):
2579 metrics.dispose(self)
2581 self._index.update(
2582 {idx: (k, col) for idx, (k, col, _) in enumerate(self._collection)}
2583 )
2584 # delete higher index
2585 del self._index[len(self._collection)]
2587 def replace(
2588 self,
2589 column: _NAMEDCOL,
2590 *,
2591 extra_remove: Optional[Iterable[_NAMEDCOL]] = None,
2592 index: Optional[int] = None,
2593 ) -> None:
2594 """add the given column to this collection, removing unaliased
2595 versions of this column as well as existing columns with the
2596 same key.
2598 e.g.::
2600 t = Table("sometable", metadata, Column("col1", Integer))
2601 t.columns.replace(Column("col1", Integer, key="columnone"))
2603 will remove the original 'col1' from the collection, and add
2604 the new column under the name 'columnname'.
2606 Used by schema.Column to override columns during table reflection.
2608 """
2610 if extra_remove:
2611 remove_col = set(extra_remove)
2612 else:
2613 remove_col = set()
2614 # remove up to two columns based on matches of name as well as key
2615 if column.name in self._index and column.key != column.name:
2616 other = self._index[column.name][1]
2617 if other.name == other.key:
2618 remove_col.add(other)
2620 if column.key in self._index:
2621 remove_col.add(self._index[column.key][1])
2623 if not remove_col:
2624 self._append_new_column(column.key, column, index=index)
2625 return
2626 new_cols: List[Tuple[str, _NAMEDCOL, _ColumnMetrics[_NAMEDCOL]]] = []
2627 replace_index = None
2629 for idx, (k, col, metrics) in enumerate(self._collection):
2630 if col in remove_col:
2631 if replace_index is None:
2632 replace_index = idx
2633 new_cols.append(
2634 (column.key, column, _ColumnMetrics(self, column))
2635 )
2636 else:
2637 new_cols.append((k, col, metrics))
2639 if remove_col:
2640 self._colset.difference_update(remove_col)
2642 for rc in remove_col:
2643 for metrics in self._proxy_index.get(rc, ()):
2644 metrics.dispose(self)
2646 if replace_index is None:
2647 if index is not None:
2648 new_cols.insert(
2649 index, (column.key, column, _ColumnMetrics(self, column))
2650 )
2652 else:
2653 new_cols.append(
2654 (column.key, column, _ColumnMetrics(self, column))
2655 )
2656 elif index is not None:
2657 to_move = new_cols[replace_index]
2658 effective_positive_index = (
2659 index if index >= 0 else max(0, len(new_cols) + index)
2660 )
2661 new_cols.insert(index, to_move)
2662 if replace_index > effective_positive_index:
2663 del new_cols[replace_index + 1]
2664 else:
2665 del new_cols[replace_index]
2667 self._colset.add(column._deannotate())
2668 self._collection[:] = new_cols
2670 self._index.clear()
2672 self._index.update(
2673 {idx: (k, col) for idx, (k, col, _) in enumerate(self._collection)}
2674 )
2675 self._index.update({k: (k, col) for (k, col, _) in self._collection})
2678class ReadOnlyColumnCollection(
2679 util.ReadOnlyContainer, ColumnCollection[_COLKEY, _COL_co]
2680):
2681 __slots__ = ("_parent",)
2683 _parent: WriteableColumnCollection[_COLKEY, _COL_co]
2685 def __init__(
2686 self, collection: WriteableColumnCollection[_COLKEY, _COL_co]
2687 ):
2688 object.__setattr__(self, "_parent", collection)
2689 object.__setattr__(self, "_index", collection._index)
2690 object.__setattr__(self, "_collection", collection._collection)
2691 object.__setattr__(self, "_colset", collection._colset)
2692 object.__setattr__(self, "_proxy_index", collection._proxy_index)
2694 def _as_readonly(self) -> ReadOnlyColumnCollection[_COLKEY, _COL_co]:
2695 return self
2697 def __getstate__(self) -> Dict[str, ColumnCollection[_COLKEY, _COL_co]]:
2698 return {"_parent": self._parent}
2700 def __setstate__(self, state: Dict[str, Any]) -> None:
2701 parent = state["_parent"]
2702 self.__init__(parent) # type: ignore[misc]
2704 def corresponding_column(
2705 self, column: _COL, require_embedded: bool = False
2706 ) -> Optional[Union[_COL, _COL_co]]:
2707 """Given a :class:`_expression.ColumnElement`, return the exported
2708 :class:`_expression.ColumnElement` object from this
2709 :class:`_expression.ColumnCollection`
2710 which corresponds to that original :class:`_expression.ColumnElement`
2711 via a common
2712 ancestor column.
2714 See :meth:`.ColumnCollection.corresponding_column` for parameter
2715 information.
2717 """
2718 return self._parent.corresponding_column(column, require_embedded)
2721class ColumnSet(util.OrderedSet["ColumnClause[Any]"]):
2722 def contains_column(self, col: ColumnClause[Any]) -> bool:
2723 return col in self
2725 def extend(self, cols: Iterable[Any]) -> None:
2726 for col in cols:
2727 self.add(col)
2729 def __eq__(self, other):
2730 l = []
2731 for c in other:
2732 for local in self:
2733 if c.shares_lineage(local):
2734 l.append(c == local)
2735 return elements.and_(*l)
2737 def __hash__(self) -> int: # type: ignore[override]
2738 return hash(tuple(x for x in self))
2741def _entity_namespace(
2742 entity: Union[_HasEntityNamespace, ExternallyTraversible],
2743) -> _EntityNamespace:
2744 """Return the nearest .entity_namespace for the given entity.
2746 If not immediately available, does an iterate to find a sub-element
2747 that has one, if any.
2749 """
2750 try:
2751 return cast(_HasEntityNamespace, entity).entity_namespace
2752 except AttributeError:
2753 for elem in visitors.iterate(cast(ExternallyTraversible, entity)):
2754 if _is_has_entity_namespace(elem):
2755 return elem.entity_namespace
2756 else:
2757 raise
2760@overload
2761def _entity_namespace_key(
2762 entity: Union[_HasEntityNamespace, ExternallyTraversible],
2763 key: str,
2764) -> SQLCoreOperations[Any]: ...
2767@overload
2768def _entity_namespace_key(
2769 entity: Union[_HasEntityNamespace, ExternallyTraversible],
2770 key: str,
2771 default: _NoArg,
2772) -> SQLCoreOperations[Any]: ...
2775@overload
2776def _entity_namespace_key(
2777 entity: Union[_HasEntityNamespace, ExternallyTraversible],
2778 key: str,
2779 default: _T,
2780) -> Union[SQLCoreOperations[Any], _T]: ...
2783def _entity_namespace_key(
2784 entity: Union[_HasEntityNamespace, ExternallyTraversible],
2785 key: str,
2786 default: Union[SQLCoreOperations[Any], _T, _NoArg] = NO_ARG,
2787) -> Union[SQLCoreOperations[Any], _T]:
2788 """Return an entry from an entity_namespace.
2791 Raises :class:`_exc.InvalidRequestError` rather than attribute error
2792 on not found.
2794 """
2796 try:
2797 ns = _entity_namespace(entity)
2798 if default is not NO_ARG:
2799 return getattr(ns, key, default)
2800 else:
2801 return getattr(ns, key) # type: ignore[no-any-return]
2802 except AttributeError as err:
2803 raise exc.InvalidRequestError(
2804 'Entity namespace for "%s" has no property "%s"' % (entity, key)
2805 ) from err
2808def _entity_namespace_key_search_all(
2809 entities: Collection[Any],
2810 key: str,
2811) -> SQLCoreOperations[Any]:
2812 """Search multiple entities for a key, raise if ambiguous or not found.
2814 This is used by filter_by() to search across all FROM clause entities
2815 when a single entity doesn't have the requested attribute.
2817 .. versionadded:: 2.1
2819 Raises:
2820 AmbiguousColumnError: If key exists in multiple entities
2821 InvalidRequestError: If key doesn't exist in any entity
2822 """
2824 match_: SQLCoreOperations[Any] | None = None
2826 for entity in entities:
2827 ns = _entity_namespace(entity)
2828 # Check if the attribute exists
2829 if hasattr(ns, key):
2830 if match_ is not None:
2831 entity_desc = ", ".join(str(e) for e in list(entities)[:3])
2832 if len(entities) > 3:
2833 entity_desc += f", ... ({len(entities)} total)"
2834 raise exc.AmbiguousColumnError(
2835 f'Attribute name "{key}" is ambiguous; it exists in '
2836 f"multiple FROM clause entities ({entity_desc}). "
2837 f"Use filter() with explicit column references instead "
2838 f"of filter_by()."
2839 )
2840 match_ = getattr(ns, key)
2842 if match_ is None:
2843 # No entity has this attribute
2844 entity_desc = ", ".join(str(e) for e in list(entities)[:3])
2845 if len(entities) > 3:
2846 entity_desc += f", ... ({len(entities)} total)"
2847 raise exc.InvalidRequestError(
2848 f'None of the FROM clause entities have a property "{key}". '
2849 f"Searched entities: {entity_desc}"
2850 )
2852 return match_