Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/sqlalchemy/sql/base.py: 48%
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 _DialectArgView(MutableMapping[str, Any]):
381 """A dictionary view of dialect-level arguments in the form
382 <dialectname>_<argument_name>.
384 """
386 __slots__ = ("obj",)
388 def __init__(self, obj: DialectKWArgs) -> None:
389 self.obj = obj
391 def _key(self, key: str) -> Tuple[str, str]:
392 try:
393 dialect, value_key = key.split("_", 1)
394 except ValueError as err:
395 raise KeyError(key) from err
396 else:
397 return dialect, value_key
399 def __getitem__(self, key: str) -> Any:
400 dialect, value_key = self._key(key)
402 try:
403 opt = self.obj.dialect_options[dialect]
404 except exc.NoSuchModuleError as err:
405 raise KeyError(key) from err
406 else:
407 return opt[value_key]
409 def __setitem__(self, key: str, value: Any) -> None:
410 try:
411 dialect, value_key = self._key(key)
412 except KeyError as err:
413 raise exc.ArgumentError(
414 "Keys must be of the form <dialectname>_<argname>"
415 ) from err
416 else:
417 self.obj.dialect_options[dialect][value_key] = value
419 def __delitem__(self, key: str) -> None:
420 dialect, value_key = self._key(key)
421 del self.obj.dialect_options[dialect][value_key]
423 def __len__(self) -> int:
424 return sum(
425 len(args._non_defaults)
426 for args in self.obj.dialect_options.values()
427 )
429 def __iter__(self) -> Generator[str, None, None]:
430 return (
431 "%s_%s" % (dialect_name, value_name)
432 for dialect_name in self.obj.dialect_options
433 for value_name in self.obj.dialect_options[
434 dialect_name
435 ]._non_defaults
436 )
439class _DialectArgDict(MutableMapping[str, Any]):
440 """A dictionary view of dialect-level arguments for a specific
441 dialect.
443 Maintains a separate collection of user-specified arguments
444 and dialect-specified default arguments.
446 """
448 def __init__(self) -> None:
449 self._non_defaults: Dict[str, Any] = {}
450 self._defaults: Dict[str, Any] = {}
452 def __len__(self) -> int:
453 return len(set(self._non_defaults).union(self._defaults))
455 def __iter__(self) -> Iterator[str]:
456 return iter(set(self._non_defaults).union(self._defaults))
458 def __getitem__(self, key: str) -> Any:
459 if key in self._non_defaults:
460 return self._non_defaults[key]
461 else:
462 return self._defaults[key]
464 def __setitem__(self, key: str, value: Any) -> None:
465 self._non_defaults[key] = value
467 def __delitem__(self, key: str) -> None:
468 del self._non_defaults[key]
471@util.preload_module("sqlalchemy.dialects")
472def _kw_reg_for_dialect(dialect_name: str) -> Optional[Dict[Any, Any]]:
473 dialect_cls = util.preloaded.dialects.registry.load(dialect_name)
474 if dialect_cls.construct_arguments is None:
475 return None
476 return dict(dialect_cls.construct_arguments)
479class DialectKWArgs:
480 """Establish the ability for a class to have dialect-specific arguments
481 with defaults and constructor validation.
483 The :class:`.DialectKWArgs` interacts with the
484 :attr:`.DefaultDialect.construct_arguments` present on a dialect.
486 .. seealso::
488 :attr:`.DefaultDialect.construct_arguments`
490 """
492 __slots__ = ()
494 _dialect_kwargs_traverse_internals: List[Tuple[str, Any]] = [
495 ("dialect_options", InternalTraversal.dp_dialect_options)
496 ]
498 def get_dialect_option(
499 self,
500 dialect: Dialect,
501 argument_name: str,
502 *,
503 else_: Any = None,
504 deprecated_fallback: Optional[str] = None,
505 ) -> Any:
506 r"""Return the value of a dialect-specific option, or *else_* if
507 this dialect does not register the given argument.
509 This is useful for DDL compilers that may be inherited by
510 third-party dialects whose ``construct_arguments`` do not
511 include the same set of keys as the parent dialect.
513 :param dialect: The dialect for which to retrieve the option.
514 :param argument_name: The name of the argument to retrieve.
515 :param else\_: The value to return if the argument is not present.
516 :param deprecated_fallback: Optional dialect name to fall back to
517 if the argument is not present for the current dialect. If the
518 argument is present for the fallback dialect but not the current
519 dialect, a deprecation warning will be emitted.
521 """
523 registry = DialectKWArgs._kw_registry[dialect.name]
524 if registry is None:
525 return else_
527 if argument_name in registry.get(self.__class__, {}):
528 if (
529 deprecated_fallback is None
530 or dialect.name == deprecated_fallback
531 ):
532 return self.dialect_options[dialect.name][argument_name]
534 # deprecated_fallback is present; need to look in two places
536 # Current dialect has this option registered.
537 # Check if user explicitly set it.
538 if (
539 dialect.name in self.dialect_options
540 and argument_name
541 in self.dialect_options[dialect.name]._non_defaults
542 ):
543 # User explicitly set this dialect's option - use it
544 return self.dialect_options[dialect.name][argument_name]
546 # User didn't set current dialect's option.
547 # Check for deprecated fallback.
548 elif (
549 deprecated_fallback in self.dialect_options
550 and argument_name
551 in self.dialect_options[deprecated_fallback]._non_defaults
552 ):
553 # User set fallback option but not current dialect's option
554 warn_deprecated(
555 f"Using '{deprecated_fallback}_{argument_name}' "
556 f"with the '{dialect.name}' dialect is deprecated; "
557 f"please additionally specify "
558 f"'{dialect.name}_{argument_name}'.",
559 version="2.1",
560 )
561 return self.dialect_options[deprecated_fallback][argument_name]
563 # Return default value
564 return self.dialect_options[dialect.name][argument_name]
565 else:
566 # Current dialect doesn't have the option registered at all.
567 # Don't warn - if a third-party dialect doesn't support an
568 # option, that's their choice, not a deprecation case.
569 return else_
571 @classmethod
572 def argument_for(
573 cls, dialect_name: str, argument_name: str, default: Any
574 ) -> None:
575 """Add a new kind of dialect-specific keyword argument for this class.
577 E.g.::
579 Index.argument_for("mydialect", "length", None)
581 some_index = Index("a", "b", mydialect_length=5)
583 The :meth:`.DialectKWArgs.argument_for` method is a per-argument
584 way adding extra arguments to the
585 :attr:`.DefaultDialect.construct_arguments` dictionary. This
586 dictionary provides a list of argument names accepted by various
587 schema-level constructs on behalf of a dialect.
589 New dialects should typically specify this dictionary all at once as a
590 data member of the dialect class. The use case for ad-hoc addition of
591 argument names is typically for end-user code that is also using
592 a custom compilation scheme which consumes the additional arguments.
594 :param dialect_name: name of a dialect. The dialect must be
595 locatable, else a :class:`.NoSuchModuleError` is raised. The
596 dialect must also include an existing
597 :attr:`.DefaultDialect.construct_arguments` collection, indicating
598 that it participates in the keyword-argument validation and default
599 system, else :class:`.ArgumentError` is raised. If the dialect does
600 not include this collection, then any keyword argument can be
601 specified on behalf of this dialect already. All dialects packaged
602 within SQLAlchemy include this collection, however for third party
603 dialects, support may vary.
605 :param argument_name: name of the parameter.
607 :param default: default value of the parameter.
609 """
611 construct_arg_dictionary: Optional[Dict[Any, Any]] = (
612 DialectKWArgs._kw_registry[dialect_name]
613 )
614 if construct_arg_dictionary is None:
615 raise exc.ArgumentError(
616 "Dialect '%s' does have keyword-argument "
617 "validation and defaults enabled configured" % dialect_name
618 )
619 if cls not in construct_arg_dictionary:
620 construct_arg_dictionary[cls] = {}
621 construct_arg_dictionary[cls][argument_name] = default
623 @property
624 def dialect_kwargs(self) -> _DialectArgView:
625 """A collection of keyword arguments specified as dialect-specific
626 options to this construct.
628 The arguments are present here in their original ``<dialect>_<kwarg>``
629 format. Only arguments that were actually passed are included;
630 unlike the :attr:`.DialectKWArgs.dialect_options` collection, which
631 contains all options known by this dialect including defaults.
633 The collection is also writable; keys are accepted of the
634 form ``<dialect>_<kwarg>`` where the value will be assembled
635 into the list of options.
637 .. seealso::
639 :attr:`.DialectKWArgs.dialect_options` - nested dictionary form
641 """
642 return _DialectArgView(self)
644 @property
645 def kwargs(self) -> _DialectArgView:
646 """A synonym for :attr:`.DialectKWArgs.dialect_kwargs`."""
647 return self.dialect_kwargs
649 _kw_registry: util.PopulateDict[str, Optional[Dict[Any, Any]]] = (
650 util.PopulateDict(_kw_reg_for_dialect)
651 )
653 @classmethod
654 def _kw_reg_for_dialect_cls(cls, dialect_name: str) -> _DialectArgDict:
655 construct_arg_dictionary = DialectKWArgs._kw_registry[dialect_name]
656 d = _DialectArgDict()
658 if construct_arg_dictionary is None:
659 d._defaults.update({"*": None})
660 else:
661 for cls in reversed(cls.__mro__):
662 if cls in construct_arg_dictionary:
663 d._defaults.update(construct_arg_dictionary[cls])
664 return d
666 @util.memoized_property
667 def dialect_options(self) -> util.PopulateDict[str, _DialectArgDict]:
668 """A collection of keyword arguments specified as dialect-specific
669 options to this construct.
671 This is a two-level nested registry, keyed to ``<dialect_name>``
672 and ``<argument_name>``. For example, the ``postgresql_where``
673 argument would be locatable as::
675 arg = my_object.dialect_options["postgresql"]["where"]
677 .. versionadded:: 0.9.2
679 .. seealso::
681 :attr:`.DialectKWArgs.dialect_kwargs` - flat dictionary form
683 """
685 return util.PopulateDict(self._kw_reg_for_dialect_cls)
687 def _validate_dialect_kwargs(self, kwargs: Dict[str, Any]) -> None:
688 # validate remaining kwargs that they all specify DB prefixes
690 if not kwargs:
691 return
693 for k in kwargs:
694 m = re.match("^(.+?)_(.+)$", k)
695 if not m:
696 raise TypeError(
697 "Additional arguments should be "
698 "named <dialectname>_<argument>, got '%s'" % k
699 )
700 dialect_name, arg_name = m.group(1, 2)
702 try:
703 construct_arg_dictionary = self.dialect_options[dialect_name]
704 except exc.NoSuchModuleError:
705 util.warn(
706 "Can't validate argument %r; can't "
707 "locate any SQLAlchemy dialect named %r"
708 % (k, dialect_name)
709 )
710 self.dialect_options[dialect_name] = d = _DialectArgDict()
711 d._defaults.update({"*": None})
712 d._non_defaults[arg_name] = kwargs[k]
713 else:
714 if (
715 "*" not in construct_arg_dictionary
716 and arg_name not in construct_arg_dictionary
717 ):
718 raise exc.ArgumentError(
719 "Argument %r is not accepted by "
720 "dialect %r on behalf of %r"
721 % (k, dialect_name, self.__class__)
722 )
723 else:
724 construct_arg_dictionary[arg_name] = kwargs[k]
727class CompileState:
728 """Produces additional object state necessary for a statement to be
729 compiled.
731 the :class:`.CompileState` class is at the base of classes that assemble
732 state for a particular statement object that is then used by the
733 compiler. This process is essentially an extension of the process that
734 the SQLCompiler.visit_XYZ() method takes, however there is an emphasis
735 on converting raw user intent into more organized structures rather than
736 producing string output. The top-level :class:`.CompileState` for the
737 statement being executed is also accessible when the execution context
738 works with invoking the statement and collecting results.
740 The production of :class:`.CompileState` is specific to the compiler, such
741 as within the :meth:`.SQLCompiler.visit_insert`,
742 :meth:`.SQLCompiler.visit_select` etc. methods. These methods are also
743 responsible for associating the :class:`.CompileState` with the
744 :class:`.SQLCompiler` itself, if the statement is the "toplevel" statement,
745 i.e. the outermost SQL statement that's actually being executed.
746 There can be other :class:`.CompileState` objects that are not the
747 toplevel, such as when a SELECT subquery or CTE-nested
748 INSERT/UPDATE/DELETE is generated.
750 .. versionadded:: 1.4
752 """
754 __slots__ = ("statement", "_ambiguous_table_name_map")
756 plugins: Dict[Tuple[str, str], Type[CompileState]] = {}
758 _ambiguous_table_name_map: Optional[_AmbiguousTableNameMap]
760 @classmethod
761 def create_for_statement(
762 cls, statement: Executable, compiler: SQLCompiler, **kw: Any
763 ) -> CompileState:
764 # factory construction.
766 if statement._propagate_attrs:
767 plugin_name = statement._propagate_attrs.get(
768 "compile_state_plugin", "default"
769 )
770 klass = cls.plugins.get(
771 (plugin_name, statement._effective_plugin_target), None
772 )
773 if klass is None:
774 klass = cls.plugins[
775 ("default", statement._effective_plugin_target)
776 ]
778 else:
779 klass = cls.plugins[
780 ("default", statement._effective_plugin_target)
781 ]
783 if klass is cls:
784 return cls(statement, compiler, **kw)
785 else:
786 return klass.create_for_statement(statement, compiler, **kw)
788 def __init__(self, statement, compiler, **kw):
789 self.statement = statement
791 @classmethod
792 def get_plugin_class(
793 cls, statement: Executable
794 ) -> Optional[Type[CompileState]]:
795 plugin_name = statement._propagate_attrs.get(
796 "compile_state_plugin", None
797 )
799 if plugin_name:
800 key = (plugin_name, statement._effective_plugin_target)
801 if key in cls.plugins:
802 return cls.plugins[key]
804 # there's no case where we call upon get_plugin_class() and want
805 # to get None back, there should always be a default. return that
806 # if there was no plugin-specific class (e.g. Insert with "orm"
807 # plugin)
808 try:
809 return cls.plugins[("default", statement._effective_plugin_target)]
810 except KeyError:
811 return None
813 @classmethod
814 def _get_plugin_class_for_plugin(
815 cls, statement: Executable, plugin_name: str
816 ) -> Optional[Type[CompileState]]:
817 try:
818 return cls.plugins[
819 (plugin_name, statement._effective_plugin_target)
820 ]
821 except KeyError:
822 return None
824 @classmethod
825 def plugin_for(
826 cls, plugin_name: str, visit_name: str
827 ) -> Callable[[_Fn], _Fn]:
828 def decorate(cls_to_decorate):
829 cls.plugins[(plugin_name, visit_name)] = cls_to_decorate
830 return cls_to_decorate
832 return decorate
835class Generative(HasMemoized):
836 """Provide a method-chaining pattern in conjunction with the
837 @_generative decorator."""
839 def _generate(self) -> Self:
840 skip = self._memoized_keys
841 cls = self.__class__
842 s = cls.__new__(cls)
843 if skip:
844 # ensure this iteration remains atomic
845 s.__dict__ = {
846 k: v for k, v in self.__dict__.copy().items() if k not in skip
847 }
848 else:
849 s.__dict__ = self.__dict__.copy()
850 return s
853class InPlaceGenerative(HasMemoized):
854 """Provide a method-chaining pattern in conjunction with the
855 @_generative decorator that mutates in place."""
857 __slots__ = ()
859 def _generate(self) -> Self:
860 skip = self._memoized_keys
861 # note __dict__ needs to be in __slots__ if this is used
862 for k in skip:
863 self.__dict__.pop(k, None)
864 return self
867class HasCompileState(Generative):
868 """A class that has a :class:`.CompileState` associated with it."""
870 _compile_state_plugin: Optional[Type[CompileState]] = None
872 _attributes: util.immutabledict[str, Any] = util.EMPTY_DICT
874 _compile_state_factory = CompileState.create_for_statement
877class _MetaOptions(type):
878 """metaclass for the Options class.
880 This metaclass is actually necessary despite the availability of the
881 ``__init_subclass__()`` hook as this type also provides custom class-level
882 behavior for the ``__add__()`` method.
884 """
886 _cache_attrs: Tuple[str, ...]
888 def __add__(self, other):
889 o1 = self()
891 if set(other).difference(self._cache_attrs):
892 raise TypeError(
893 "dictionary contains attributes not covered by "
894 "Options class %s: %r"
895 % (self, set(other).difference(self._cache_attrs))
896 )
898 o1.__dict__.update(other)
899 return o1
901 if TYPE_CHECKING:
903 def __getattr__(self, key: str) -> Any: ...
905 def __setattr__(self, key: str, value: Any) -> None: ...
907 def __delattr__(self, key: str) -> None: ...
910class Options(metaclass=_MetaOptions):
911 """A cacheable option dictionary with defaults."""
913 __slots__ = ()
915 _cache_attrs: Tuple[str, ...]
917 def __init_subclass__(cls) -> None:
918 dict_ = cls.__dict__
919 cls._cache_attrs = tuple(
920 sorted(
921 d
922 for d in dict_
923 if not d.startswith("__")
924 and d not in ("_cache_key_traversal",)
925 )
926 )
927 super().__init_subclass__()
929 def __init__(self, **kw: Any) -> None:
930 self.__dict__.update(kw)
932 def __add__(self, other):
933 o1 = self.__class__.__new__(self.__class__)
934 o1.__dict__.update(self.__dict__)
936 if set(other).difference(self._cache_attrs):
937 raise TypeError(
938 "dictionary contains attributes not covered by "
939 "Options class %s: %r"
940 % (self, set(other).difference(self._cache_attrs))
941 )
943 o1.__dict__.update(other)
944 return o1
946 def __eq__(self, other):
947 # TODO: very inefficient. This is used only in test suites
948 # right now.
949 for a, b in zip_longest(self._cache_attrs, other._cache_attrs):
950 if getattr(self, a) != getattr(other, b):
951 return False
952 return True
954 def __repr__(self) -> str:
955 # TODO: fairly inefficient, used only in debugging right now.
957 return "%s(%s)" % (
958 self.__class__.__name__,
959 ", ".join(
960 "%s=%r" % (k, self.__dict__[k])
961 for k in self._cache_attrs
962 if k in self.__dict__
963 ),
964 )
966 @classmethod
967 def isinstance(cls, klass: Type[Any]) -> bool:
968 return issubclass(cls, klass)
970 @hybridmethod
971 def add_to_element(self, name: str, value: str) -> Any:
972 return self + {name: getattr(self, name) + value}
974 @hybridmethod
975 def _state_dict_inst(self) -> Mapping[str, Any]:
976 return self.__dict__
978 _state_dict_const: util.immutabledict[str, Any] = util.EMPTY_DICT
980 @_state_dict_inst.classlevel
981 def _state_dict(cls) -> Mapping[str, Any]:
982 return cls._state_dict_const
984 @classmethod
985 def safe_merge(cls, other: "Options") -> Any:
986 d = other._state_dict()
988 # only support a merge with another object of our class
989 # and which does not have attrs that we don't. otherwise
990 # we risk having state that might not be part of our cache
991 # key strategy
993 if (
994 cls is not other.__class__
995 and other._cache_attrs
996 and set(other._cache_attrs).difference(cls._cache_attrs)
997 ):
998 raise TypeError(
999 "other element %r is not empty, is not of type %s, "
1000 "and contains attributes not covered here %r"
1001 % (
1002 other,
1003 cls,
1004 set(other._cache_attrs).difference(cls._cache_attrs),
1005 )
1006 )
1007 return cls + d
1009 @classmethod
1010 def from_execution_options(
1011 cls,
1012 key: str,
1013 attrs: set[str],
1014 exec_options: Mapping[str, Any],
1015 statement_exec_options: Mapping[str, Any],
1016 ) -> Tuple["Options", Mapping[str, Any]]:
1017 """process Options argument in terms of execution options.
1020 e.g.::
1022 (
1023 load_options,
1024 execution_options,
1025 ) = QueryContext.default_load_options.from_execution_options(
1026 "_sa_orm_load_options",
1027 {"populate_existing", "autoflush", "yield_per"},
1028 execution_options,
1029 statement._execution_options,
1030 )
1032 get back the Options and refresh "_sa_orm_load_options" in the
1033 exec options dict w/ the Options as well
1035 """
1037 # common case is that no options we are looking for are
1038 # in either dictionary, so cancel for that first
1039 check_argnames = attrs.intersection(
1040 set(exec_options).union(statement_exec_options)
1041 )
1043 existing_options = exec_options.get(key, cls)
1045 if check_argnames:
1046 result = {}
1047 for argname in check_argnames:
1048 local = "_" + argname
1049 if argname in exec_options:
1050 result[local] = exec_options[argname]
1051 elif argname in statement_exec_options:
1052 result[local] = statement_exec_options[argname]
1054 new_options = existing_options + result
1055 exec_options = util.EMPTY_DICT.merge_with(
1056 exec_options, {key: new_options}
1057 )
1058 return new_options, exec_options
1060 else:
1061 return existing_options, exec_options
1063 if TYPE_CHECKING:
1065 def __getattr__(self, key: str) -> Any: ...
1067 def __setattr__(self, key: str, value: Any) -> None: ...
1069 def __delattr__(self, key: str) -> None: ...
1072class CacheableOptions(Options, HasCacheKey):
1073 __slots__ = ()
1075 @hybridmethod
1076 def _gen_cache_key_inst(
1077 self, anon_map: Any, bindparams: List[BindParameter[Any]]
1078 ) -> Optional[Tuple[Any]]:
1079 # _gen_cache_key is a compiled function in _cache_key_cy; its
1080 # cython directives make mypy see it as untyped
1081 return HasCacheKey._gen_cache_key( # type: ignore[no-any-return] # noqa: E501
1082 self, anon_map, bindparams
1083 )
1085 @_gen_cache_key_inst.classlevel
1086 def _gen_cache_key(
1087 cls, anon_map: "anon_map", bindparams: List[BindParameter[Any]]
1088 ) -> Tuple[CacheableOptions, Any]:
1089 return (cls, ())
1091 @hybridmethod
1092 def _generate_cache_key(self) -> Optional[CacheKey]:
1093 return HasCacheKey._generate_cache_key(self)
1096class ExecutableOption(HasCopyInternals):
1097 __slots__ = ()
1099 _annotations: _ImmutableExecuteOptions = util.EMPTY_DICT
1101 __visit_name__: str = "executable_option"
1103 _is_has_cache_key: bool = False
1105 _is_core: bool = True
1107 def _clone(self, **kw):
1108 """Create a shallow copy of this ExecutableOption."""
1109 c = self.__class__.__new__(self.__class__)
1110 c.__dict__ = dict(self.__dict__) # type: ignore[misc]
1111 return c
1114_L = TypeVar("_L", bound=str)
1117class HasSyntaxExtensions(Generic[_L]):
1119 _position_map: Mapping[_L, str]
1121 @_generative
1122 def ext(self, extension: SyntaxExtension) -> Self:
1123 """Applies a SQL syntax extension to this statement.
1125 SQL syntax extensions are :class:`.ClauseElement` objects that define
1126 some vendor-specific syntactical construct that take place in specific
1127 parts of a SQL statement. Examples include vendor extensions like
1128 PostgreSQL / SQLite's "ON DUPLICATE KEY UPDATE", PostgreSQL's
1129 "DISTINCT ON", and MySQL's "LIMIT" that can be applied to UPDATE
1130 and DELETE statements.
1132 .. seealso::
1134 :ref:`examples_syntax_extensions`
1136 :func:`_mysql.limit` - DML LIMIT for MySQL
1138 :func:`_postgresql.distinct_on` - DISTINCT ON for PostgreSQL
1140 .. versionadded:: 2.1
1142 """
1143 extension = coercions.expect(
1144 roles.SyntaxExtensionRole, extension, apply_propagate_attrs=self
1145 )
1146 self._apply_syntax_extension_to_self(extension)
1147 return self
1149 @util.preload_module("sqlalchemy.sql.elements")
1150 def apply_syntax_extension_point(
1151 self,
1152 apply_fn: Callable[[Sequence[ClauseElement]], Sequence[ClauseElement]],
1153 position: _L,
1154 ) -> None:
1155 """Apply a :class:`.SyntaxExtension` to a known extension point.
1157 Should be used only internally by :class:`.SyntaxExtension`.
1159 E.g.::
1161 class Qualify(SyntaxExtension, ClauseElement):
1163 # ...
1165 def apply_to_select(self, select_stmt: Select) -> None:
1166 # append self to existing
1167 select_stmt.apply_extension_point(
1168 lambda existing: [*existing, self], "post_criteria"
1169 )
1172 class ReplaceExt(SyntaxExtension, ClauseElement):
1174 # ...
1176 def apply_to_select(self, select_stmt: Select) -> None:
1177 # replace any existing elements regardless of type
1178 select_stmt.apply_extension_point(
1179 lambda existing: [self], "post_criteria"
1180 )
1183 class ReplaceOfTypeExt(SyntaxExtension, ClauseElement):
1185 # ...
1187 def apply_to_select(self, select_stmt: Select) -> None:
1188 # replace any existing elements of the same type
1189 select_stmt.apply_extension_point(
1190 self.append_replacing_same_type, "post_criteria"
1191 )
1193 :param apply_fn: callable function that will receive a sequence of
1194 :class:`.ClauseElement` that is already populating the extension
1195 point (the sequence is empty if there isn't one), and should return
1196 a new sequence of :class:`.ClauseElement` that will newly populate
1197 that point. The function typically can choose to concatenate the
1198 existing values with the new one, or to replace the values that are
1199 there with a new one by returning a list of a single element, or
1200 to perform more complex operations like removing only the same
1201 type element from the input list of merging already existing elements
1202 of the same type. Some examples are shown in the examples above
1203 :param position: string name of the position to apply to. This
1204 varies per statement type. IDEs should show the possible values
1205 for each statement type as it's typed with a ``typing.Literal`` per
1206 statement.
1208 .. seealso::
1210 :ref:`examples_syntax_extensions`
1212 :meth:`.ext`
1215 """ # noqa: E501
1217 try:
1218 attrname = self._position_map[position]
1219 except KeyError as ke:
1220 raise ValueError(
1221 f"Unknown position {position!r} for {self.__class__} "
1222 f"construct; known positions: "
1223 f"{', '.join(repr(k) for k in self._position_map)}"
1224 ) from ke
1225 else:
1226 ElementList = util.preloaded.sql_elements.ElementList
1227 existing: Optional[ClauseElement] = getattr(self, attrname, None)
1228 if existing is None:
1229 input_seq: Tuple[ClauseElement, ...] = ()
1230 elif isinstance(existing, ElementList):
1231 input_seq = existing.clauses
1232 else:
1233 input_seq = (existing,)
1235 new_seq = apply_fn(input_seq)
1236 assert new_seq, "cannot return empty sequence"
1237 new = new_seq[0] if len(new_seq) == 1 else ElementList(new_seq)
1238 setattr(self, attrname, new)
1240 def _apply_syntax_extension_to_self(
1241 self, extension: SyntaxExtension
1242 ) -> None:
1243 raise NotImplementedError()
1245 def _get_syntax_extensions_as_dict(self) -> Mapping[_L, SyntaxExtension]:
1246 res: Dict[_L, SyntaxExtension] = {}
1247 for name, attr in self._position_map.items():
1248 value = getattr(self, attr)
1249 if value is not None:
1250 res[name] = value
1251 return res
1253 def _set_syntax_extensions(self, **extensions: SyntaxExtension) -> None:
1254 for name, value in extensions.items():
1255 setattr(self, self._position_map[name], value) # type: ignore[index] # noqa: E501
1258class SyntaxExtension(roles.SyntaxExtensionRole):
1259 """Defines a unit that when also extending from :class:`.ClauseElement`
1260 can be applied to SQLAlchemy statements :class:`.Select`,
1261 :class:`_sql.Insert`, :class:`.Update` and :class:`.Delete` making use of
1262 pre-established SQL insertion points within these constructs.
1264 .. versionadded:: 2.1
1266 .. seealso::
1268 :ref:`examples_syntax_extensions`
1270 """
1272 def append_replacing_same_type(
1273 self, existing: Sequence[ClauseElement]
1274 ) -> Sequence[ClauseElement]:
1275 """Utility function that can be used as
1276 :paramref:`_sql.Select.apply_syntax_extension_point.apply_fn`
1277 to remove any other element of the same type in existing and appending
1278 ``self`` to the list.
1280 This is equivalent to::
1282 stmt.apply_syntax_extension_point(
1283 lambda existing: [
1284 *(e for e in existing if not isinstance(e, ReplaceOfTypeExt)),
1285 self,
1286 ],
1287 "post_criteria",
1288 )
1290 .. seealso::
1292 :ref:`examples_syntax_extensions`
1294 :meth:`_sql.Select.apply_syntax_extension_point` and equivalents
1295 in :class:`_dml.Insert`, :class:`_dml.Delete`, :class:`_dml.Update`
1297 """ # noqa: E501
1298 cls = type(self)
1299 return [*(e for e in existing if not isinstance(e, cls)), self] # type: ignore[list-item] # noqa: E501
1301 def apply_to_select(self, select_stmt: Select[Unpack[_Ts]]) -> None:
1302 """Apply this :class:`.SyntaxExtension` to a :class:`.Select`"""
1303 raise NotImplementedError(
1304 f"Extension {type(self).__name__} cannot be applied to select"
1305 )
1307 def apply_to_update(self, update_stmt: Update) -> None:
1308 """Apply this :class:`.SyntaxExtension` to an :class:`.Update`"""
1309 raise NotImplementedError(
1310 f"Extension {type(self).__name__} cannot be applied to update"
1311 )
1313 def apply_to_delete(self, delete_stmt: Delete) -> None:
1314 """Apply this :class:`.SyntaxExtension` to a :class:`.Delete`"""
1315 raise NotImplementedError(
1316 f"Extension {type(self).__name__} cannot be applied to delete"
1317 )
1319 def apply_to_insert(self, insert_stmt: Insert) -> None:
1320 """Apply this :class:`.SyntaxExtension` to an :class:`_sql.Insert`"""
1321 raise NotImplementedError(
1322 f"Extension {type(self).__name__} cannot be applied to insert"
1323 )
1326class Executable(roles.StatementRole):
1327 """Mark a :class:`_expression.ClauseElement` as supporting execution.
1329 :class:`.Executable` is a superclass for all "statement" types
1330 of objects, including :func:`select`, :func:`delete`, :func:`update`,
1331 :func:`insert`, :func:`text`.
1333 """
1335 supports_execution: bool = True
1336 _execution_options: _ImmutableExecuteOptions = util.EMPTY_DICT
1337 _is_default_generator: bool = False
1338 _with_options: Tuple[ExecutableOption, ...] = ()
1339 _compile_state_funcs: Tuple[
1340 Tuple[Callable[[CompileState], None], Any], ...
1341 ] = ()
1342 _compile_options: Optional[Union[Type[CacheableOptions], CacheableOptions]]
1344 _executable_traverse_internals = [
1345 ("_with_options", InternalTraversal.dp_executable_options),
1346 (
1347 "_compile_state_funcs",
1348 ExtendedInternalTraversal.dp_compile_state_funcs,
1349 ),
1350 ("_propagate_attrs", ExtendedInternalTraversal.dp_propagate_attrs),
1351 ]
1353 is_select: bool = False
1354 is_from_statement: bool = False
1355 is_update: bool = False
1356 is_insert: bool = False
1357 is_text: bool = False
1358 is_delete: bool = False
1359 is_dml: bool = False
1361 if TYPE_CHECKING:
1362 __visit_name__: str
1364 def _compile_w_cache(
1365 self,
1366 dialect: Dialect,
1367 *,
1368 compiled_cache: Optional[CompiledCacheType],
1369 column_keys: List[str],
1370 for_executemany: bool = False,
1371 schema_translate_map: Optional[SchemaTranslateMapType] = None,
1372 **kw: Any,
1373 ) -> tuple[
1374 Compiled,
1375 Sequence[BindParameter[Any]] | None,
1376 _CoreSingleExecuteParams | None,
1377 CacheStats,
1378 ]: ...
1380 def _execute_on_connection(
1381 self,
1382 connection: Connection,
1383 distilled_params: _CoreMultiExecuteParams,
1384 execution_options: CoreExecuteOptionsParameter,
1385 ) -> CursorResult[Any]: ...
1387 def _execute_on_scalar(
1388 self,
1389 connection: Connection,
1390 distilled_params: _CoreMultiExecuteParams,
1391 execution_options: CoreExecuteOptionsParameter,
1392 ) -> Any: ...
1394 @util.ro_non_memoized_property
1395 def _all_selected_columns(self) -> _SelectIterable:
1396 raise NotImplementedError()
1398 @property
1399 def _effective_plugin_target(self) -> str:
1400 return self.__visit_name__
1402 @_generative
1403 def options(self, *options: ExecutableOption) -> Self:
1404 """Apply options to this statement.
1406 In the general sense, options are any kind of Python object
1407 that can be interpreted by systems that consume the statement outside
1408 of the regular SQL compiler chain. Specifically, these options are
1409 the ORM level options that apply "eager load" and other loading
1410 behaviors to an ORM query.
1412 For background on specific kinds of options for specific kinds of
1413 statements, refer to the documentation for those option objects.
1415 .. versionchanged:: 1.4 - added :meth:`.Executable.options` to
1416 Core statement objects towards the goal of allowing unified
1417 Core / ORM querying capabilities.
1419 .. seealso::
1421 :ref:`loading_columns` - refers to options specific to the usage
1422 of ORM queries
1424 :ref:`relationship_loader_options` - refers to options specific
1425 to the usage of ORM queries
1427 """
1428 self._with_options += tuple(
1429 coercions.expect(roles.ExecutableOptionRole, opt)
1430 for opt in options
1431 )
1432 return self
1434 @_generative
1435 def _set_compile_options(self, compile_options: CacheableOptions) -> Self:
1436 """Assign the compile options to a new value.
1438 :param compile_options: appropriate CacheableOptions structure
1440 """
1442 self._compile_options = compile_options
1443 return self
1445 @_generative
1446 def _update_compile_options(self, options: CacheableOptions) -> Self:
1447 """update the _compile_options with new keys."""
1449 assert self._compile_options is not None
1450 self._compile_options += options
1451 return self
1453 @_generative
1454 def _add_compile_state_func(
1455 self,
1456 callable_: Callable[[CompileState], None],
1457 cache_args: Any,
1458 ) -> Self:
1459 """Add a compile state function to this statement.
1461 When using the ORM only, these are callable functions that will
1462 be given the CompileState object upon compilation.
1464 A second argument cache_args is required, which will be combined with
1465 the ``__code__`` identity of the function itself in order to produce a
1466 cache key.
1468 """
1469 self._compile_state_funcs += ((callable_, cache_args),)
1470 return self
1472 @overload
1473 def execution_options(
1474 self,
1475 *,
1476 compiled_cache: Optional[CompiledCacheType] = ...,
1477 logging_token: str = ...,
1478 isolation_level: IsolationLevel = ...,
1479 no_parameters: bool = False,
1480 stream_results: bool = False,
1481 max_row_buffer: int = ...,
1482 yield_per: int = ...,
1483 driver_column_names: bool = ...,
1484 insertmanyvalues_page_size: int = ...,
1485 schema_translate_map: Optional[SchemaTranslateMapType] = ...,
1486 populate_existing: bool = False,
1487 autoflush: bool = False,
1488 synchronize_session: SynchronizeSessionArgument = ...,
1489 dml_strategy: DMLStrategyArgument = ...,
1490 render_nulls: bool = ...,
1491 is_delete_using: bool = ...,
1492 is_update_from: bool = ...,
1493 preserve_rowcount: bool = False,
1494 **opt: Any,
1495 ) -> Self: ...
1497 @overload
1498 def execution_options(self, **opt: Any) -> Self: ...
1500 @_generative
1501 def execution_options(self, **kw: Any) -> Self:
1502 """Set non-SQL options for the statement which take effect during
1503 execution.
1505 Execution options can be set at many scopes, including per-statement,
1506 per-connection, or per execution, using methods such as
1507 :meth:`_engine.Connection.execution_options` and parameters which
1508 accept a dictionary of options such as
1509 :paramref:`_engine.Connection.execute.execution_options` and
1510 :paramref:`_orm.Session.execute.execution_options`.
1512 The primary characteristic of an execution option, as opposed to
1513 other kinds of options such as ORM loader options, is that
1514 **execution options never affect the compiled SQL of a query, only
1515 things that affect how the SQL statement itself is invoked or how
1516 results are fetched**. That is, execution options are not part of
1517 what's accommodated by SQL compilation nor are they considered part of
1518 the cached state of a statement.
1520 The :meth:`_sql.Executable.execution_options` method is
1521 :term:`generative`, as
1522 is the case for the method as applied to the :class:`_engine.Engine`
1523 and :class:`_orm.Query` objects, which means when the method is called,
1524 a copy of the object is returned, which applies the given parameters to
1525 that new copy, but leaves the original unchanged::
1527 statement = select(table.c.x, table.c.y)
1528 new_statement = statement.execution_options(my_option=True)
1530 An exception to this behavior is the :class:`_engine.Connection`
1531 object, where the :meth:`_engine.Connection.execution_options` method
1532 is explicitly **not** generative.
1534 The kinds of options that may be passed to
1535 :meth:`_sql.Executable.execution_options` and other related methods and
1536 parameter dictionaries include parameters that are explicitly consumed
1537 by SQLAlchemy Core or ORM, as well as arbitrary keyword arguments not
1538 defined by SQLAlchemy, which means the methods and/or parameter
1539 dictionaries may be used for user-defined parameters that interact with
1540 custom code, which may access the parameters using methods such as
1541 :meth:`_sql.Executable.get_execution_options` and
1542 :meth:`_engine.Connection.get_execution_options`, or within selected
1543 event hooks using a dedicated ``execution_options`` event parameter
1544 such as
1545 :paramref:`_events.ConnectionEvents.before_execute.execution_options`
1546 or :attr:`_orm.ORMExecuteState.execution_options`, e.g.::
1548 from sqlalchemy import event
1551 @event.listens_for(some_engine, "before_execute")
1552 def _process_opt(conn, statement, multiparams, params, execution_options):
1553 "run a SQL function before invoking a statement"
1555 if execution_options.get("do_special_thing", False):
1556 conn.exec_driver_sql("run_special_function()")
1558 Within the scope of options that are explicitly recognized by
1559 SQLAlchemy, most apply to specific classes of objects and not others.
1560 The most common execution options include:
1562 * :paramref:`_engine.Connection.execution_options.isolation_level` -
1563 sets the isolation level for a connection or a class of connections
1564 via an :class:`_engine.Engine`. This option is accepted only
1565 by :class:`_engine.Connection` or :class:`_engine.Engine`.
1567 * :paramref:`_engine.Connection.execution_options.stream_results` -
1568 indicates results should be fetched using a server side cursor;
1569 this option is accepted by :class:`_engine.Connection`, by the
1570 :paramref:`_engine.Connection.execute.execution_options` parameter
1571 on :meth:`_engine.Connection.execute`, and additionally by
1572 :meth:`_sql.Executable.execution_options` on a SQL statement object,
1573 as well as by ORM constructs like :meth:`_orm.Session.execute`.
1575 * :paramref:`_engine.Connection.execution_options.compiled_cache` -
1576 indicates a dictionary that will serve as the
1577 :ref:`SQL compilation cache <sql_caching>`
1578 for a :class:`_engine.Connection` or :class:`_engine.Engine`, as
1579 well as for ORM methods like :meth:`_orm.Session.execute`.
1580 Can be passed as ``None`` to disable caching for statements.
1581 This option is not accepted by
1582 :meth:`_sql.Executable.execution_options` as it is inadvisable to
1583 carry along a compilation cache within a statement object.
1585 * :paramref:`_engine.Connection.execution_options.schema_translate_map`
1586 - a mapping of schema names used by the
1587 :ref:`Schema Translate Map <schema_translating>` feature, accepted
1588 by :class:`_engine.Connection`, :class:`_engine.Engine`,
1589 :class:`_sql.Executable`, as well as by ORM constructs
1590 like :meth:`_orm.Session.execute`.
1592 .. seealso::
1594 :meth:`_engine.Connection.execution_options`
1596 :paramref:`_engine.Connection.execute.execution_options`
1598 :paramref:`_orm.Session.execute.execution_options`
1600 :ref:`orm_queryguide_execution_options` - documentation on all
1601 ORM-specific execution options
1603 """ # noqa: E501
1604 if "isolation_level" in kw:
1605 raise exc.ArgumentError(
1606 "'isolation_level' execution option may only be specified "
1607 "on Connection.execution_options(), or "
1608 "per-engine using the isolation_level "
1609 "argument to create_engine()."
1610 )
1611 if "compiled_cache" in kw:
1612 raise exc.ArgumentError(
1613 "'compiled_cache' execution option may only be specified "
1614 "on Connection.execution_options(), not per statement."
1615 )
1616 self._execution_options = self._execution_options.union(kw)
1617 return self
1619 def get_execution_options(self) -> _ExecuteOptions:
1620 """Get the non-SQL options which will take effect during execution.
1622 .. seealso::
1624 :meth:`.Executable.execution_options`
1625 """
1626 return self._execution_options
1629class ExecutableStatement(Executable):
1630 """Executable subclass that implements a lightweight version of ``params``
1631 that avoids a full cloned traverse.
1633 .. versionadded:: 2.1
1635 """
1637 _params: util.immutabledict[str, Any] = EMPTY_DICT
1639 _executable_traverse_internals = (
1640 Executable._executable_traverse_internals
1641 + [("_params", InternalTraversal.dp_params)]
1642 )
1644 @_generative
1645 def params(
1646 self,
1647 __optionaldict: _CoreSingleExecuteParams | None = None,
1648 /,
1649 **kwargs: Any,
1650 ) -> Self:
1651 """Return a copy with the provided bindparam values.
1653 Returns a copy of this Executable with bindparam values set
1654 to the given dictionary::
1656 >>> clause = column("x") + bindparam("foo")
1657 >>> print(clause.compile().params)
1658 {'foo': None}
1659 >>> print(clause.params({"foo": 7}).compile().params)
1660 {'foo': 7}
1662 """
1663 if __optionaldict:
1664 kwargs.update(__optionaldict)
1665 self._params = (
1666 util.immutabledict(kwargs)
1667 if not self._params
1668 else self._params | kwargs
1669 )
1670 return self
1673class SchemaEventTarget(event.EventTarget):
1674 """Base class for elements that are the targets of :class:`.DDLEvents`
1675 events.
1677 This includes :class:`.SchemaItem` as well as :class:`.SchemaType`.
1679 """
1681 dispatch: dispatcher[SchemaEventTarget]
1683 def _set_parent(self, parent: SchemaEventTarget, **kw: Any) -> None:
1684 """Associate with this SchemaEvent's parent object."""
1686 def _set_parent_with_dispatch(
1687 self, parent: SchemaEventTarget, **kw: Any
1688 ) -> None:
1689 self.dispatch.before_parent_attach(self, parent)
1690 self._set_parent(parent, **kw)
1691 self.dispatch.after_parent_attach(self, parent)
1694class SchemaVisitable(SchemaEventTarget, visitors.Visitable):
1695 """Base class for elements that are targets of a :class:`.SchemaVisitor`.
1697 .. versionadded:: 2.0.41
1699 """
1702class SchemaVisitor(ClauseVisitor):
1703 """Define the visiting for ``SchemaItem`` and more
1704 generally ``SchemaVisitable`` objects.
1706 """
1708 __traverse_options__: Dict[str, Any] = {"schema_visitor": True}
1711class _SentinelDefaultCharacterization(Enum):
1712 NONE = "none"
1713 UNKNOWN = "unknown"
1714 CLIENTSIDE = "clientside"
1715 SENTINEL_DEFAULT = "sentinel_default"
1716 SERVERSIDE = "serverside"
1717 IDENTITY = "identity"
1718 SEQUENCE = "sequence"
1719 MONOTONIC_FUNCTION = "monotonic"
1722class _SentinelColumnCharacterization(NamedTuple):
1723 columns: Optional[Sequence[Column[Any]]] = None
1724 is_explicit: bool = False
1725 is_autoinc: bool = False
1726 default_characterization: _SentinelDefaultCharacterization = (
1727 _SentinelDefaultCharacterization.NONE
1728 )
1731_COLKEY = TypeVar("_COLKEY", Union[None, str], str)
1733_COL_co = TypeVar("_COL_co", bound="ColumnElement[Any]", covariant=True)
1734_COL = TypeVar("_COL", bound="ColumnElement[Any]")
1737class _ColumnMetrics(Generic[_COL_co]):
1738 __slots__ = ("column",)
1740 column: _COL_co
1742 def __init__(
1743 self, collection: ColumnCollection[Any, _COL_co], col: _COL_co
1744 ) -> None:
1745 self.column = col
1747 # proxy_index being non-empty means it was initialized.
1748 # so we need to update it
1749 pi = collection._proxy_index
1750 if pi:
1751 for eps_col in col._expanded_proxy_set:
1752 pi[eps_col].add(self)
1754 def get_expanded_proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
1755 return self.column._expanded_proxy_set
1757 def dispose(self, collection: ColumnCollection[_COLKEY, _COL_co]) -> None:
1758 pi = collection._proxy_index
1759 if not pi:
1760 return
1761 for col in self.column._expanded_proxy_set:
1762 colset = pi.get(col, None)
1763 if colset:
1764 colset.discard(self)
1765 if colset is not None and not colset:
1766 del pi[col]
1768 def embedded(
1769 self,
1770 target_set: Union[
1771 Set[ColumnElement[Any]], FrozenSet[ColumnElement[Any]]
1772 ],
1773 ) -> bool:
1774 expanded_proxy_set = self.column._expanded_proxy_set
1775 for t in target_set.difference(expanded_proxy_set):
1776 if not expanded_proxy_set.intersection(_expand_cloned([t])):
1777 return False
1778 return True
1781class ColumnCollection(Generic[_COLKEY, _COL_co]):
1782 """Base class for collection of :class:`_expression.ColumnElement`
1783 instances, typically for :class:`_sql.FromClause` objects.
1785 The :class:`_sql.ColumnCollection` object is most commonly available
1786 as the :attr:`_schema.Table.c` or :attr:`_schema.Table.columns` collection
1787 on the :class:`_schema.Table` object, introduced at
1788 :ref:`metadata_tables_and_columns`.
1790 The :class:`_expression.ColumnCollection` has both mapping- and sequence-
1791 like behaviors. A :class:`_expression.ColumnCollection` usually stores
1792 :class:`_schema.Column` objects, which are then accessible both via mapping
1793 style access as well as attribute access style.
1795 To access :class:`_schema.Column` objects using ordinary attribute-style
1796 access, specify the name like any other object attribute, such as below
1797 a column named ``employee_name`` is accessed::
1799 >>> employee_table.c.employee_name
1801 To access columns that have names with special characters or spaces,
1802 index-style access is used, such as below which illustrates a column named
1803 ``employee ' payment`` is accessed::
1805 >>> employee_table.c["employee ' payment"]
1807 As the :class:`_sql.ColumnCollection` object provides a Python dictionary
1808 interface, common dictionary method names like
1809 :meth:`_sql.ColumnCollection.keys`, :meth:`_sql.ColumnCollection.values`,
1810 and :meth:`_sql.ColumnCollection.items` are available, which means that
1811 database columns that are keyed under these names also need to use indexed
1812 access::
1814 >>> employee_table.c["values"]
1817 The name for which a :class:`_schema.Column` would be present is normally
1818 that of the :paramref:`_schema.Column.key` parameter. In some contexts,
1819 such as a :class:`_sql.Select` object that uses a label style set
1820 using the :meth:`_sql.Select.set_label_style` method, a column of a certain
1821 key may instead be represented under a particular label name such
1822 as ``tablename_columnname``::
1824 >>> from sqlalchemy import select, column, table
1825 >>> from sqlalchemy import LABEL_STYLE_TABLENAME_PLUS_COL
1826 >>> t = table("t", column("c"))
1827 >>> stmt = select(t).set_label_style(LABEL_STYLE_TABLENAME_PLUS_COL)
1828 >>> subq = stmt.subquery()
1829 >>> subq.c.t_c
1830 <sqlalchemy.sql.elements.ColumnClause at 0x7f59dcf04fa0; t_c>
1832 :class:`.ColumnCollection` also indexes the columns in order and allows
1833 them to be accessible by their integer position::
1835 >>> cc[0]
1836 Column('x', Integer(), table=None)
1837 >>> cc[1]
1838 Column('y', Integer(), table=None)
1840 .. versionadded:: 1.4 :class:`_expression.ColumnCollection`
1841 allows integer-based
1842 index access to the collection.
1844 Iterating the collection yields the column expressions in order::
1846 >>> list(cc)
1847 [Column('x', Integer(), table=None),
1848 Column('y', Integer(), table=None)]
1850 The :class:`_expression.ColumnCollection` base class is read-only.
1851 For mutation operations, the :class:`.WriteableColumnCollection` subclass
1852 provides methods such as :meth:`.WriteableColumnCollection.add`.
1853 A special subclass :class:`.DedupeColumnCollection` exists which
1854 maintains SQLAlchemy's older behavior of not allowing duplicates; this
1855 collection is used for schema level objects like :class:`_schema.Table`
1856 and :class:`.PrimaryKeyConstraint` where this deduping is helpful.
1857 The :class:`.DedupeColumnCollection` class also has additional mutation
1858 methods as the schema constructs have more use cases that require removal
1859 and replacement of columns.
1861 .. versionchanged:: 1.4 :class:`_expression.ColumnCollection`
1862 now stores duplicate
1863 column keys as well as the same column in multiple positions. The
1864 :class:`.DedupeColumnCollection` class is added to maintain the
1865 former behavior in those cases where deduplication as well as
1866 additional replace/remove operations are needed.
1868 .. versionchanged:: 2.1 :class:`_expression.ColumnCollection` is now
1869 a read-only base class. Mutation operations are available through
1870 :class:`.WriteableColumnCollection` and :class:`.DedupeColumnCollection`
1871 subclasses.
1874 """
1876 __slots__ = ("_collection", "_index", "_colset", "_proxy_index")
1878 _collection: List[Tuple[_COLKEY, _COL_co, _ColumnMetrics[_COL_co]]]
1879 _index: Dict[Union[None, str, int], Tuple[_COLKEY, _COL_co]]
1880 _colset: Set[_COL_co]
1881 _proxy_index: Dict[ColumnElement[Any], Set[_ColumnMetrics[_COL_co]]]
1883 def __init__(self) -> None:
1884 raise TypeError(
1885 "ColumnCollection is an abstract base class and cannot be "
1886 "instantiated directly. Use WriteableColumnCollection or "
1887 "DedupeColumnCollection instead."
1888 )
1890 @util.preload_module("sqlalchemy.sql.elements")
1891 def __clause_element__(self) -> ClauseList:
1892 elements = util.preloaded.sql_elements
1894 return elements.ClauseList(
1895 _literal_as_text_role=roles.ColumnsClauseRole,
1896 group=False,
1897 *self._all_columns,
1898 )
1900 @property
1901 def _all_columns(self) -> List[_COL_co]:
1902 return [col for (_, col, _) in self._collection]
1904 def keys(self) -> List[_COLKEY]:
1905 """Return a sequence of string key names for all columns in this
1906 collection."""
1907 return [k for (k, _, _) in self._collection]
1909 def values(self) -> List[_COL_co]:
1910 """Return a sequence of :class:`_sql.ColumnClause` or
1911 :class:`_schema.Column` objects for all columns in this
1912 collection."""
1913 return [col for (_, col, _) in self._collection]
1915 def items(self) -> List[Tuple[_COLKEY, _COL_co]]:
1916 """Return a sequence of (key, column) tuples for all columns in this
1917 collection each consisting of a string key name and a
1918 :class:`_sql.ColumnClause` or
1919 :class:`_schema.Column` object.
1920 """
1922 return [(k, col) for (k, col, _) in self._collection]
1924 def __bool__(self) -> bool:
1925 return bool(self._collection)
1927 def __len__(self) -> int:
1928 return len(self._collection)
1930 def __iter__(self) -> Iterator[_COL_co]:
1931 # turn to a list first to maintain over a course of changes
1932 return iter([col for _, col, _ in self._collection])
1934 @overload
1935 def __getitem__(self, key: Union[str, int]) -> _COL_co: ...
1937 @overload
1938 def __getitem__(
1939 self, key: Union[Tuple[Union[str, int], ...], slice]
1940 ) -> ReadOnlyColumnCollection[_COLKEY, _COL_co]: ...
1942 def __getitem__(
1943 self, key: Union[str, int, slice, Tuple[Union[str, int], ...]]
1944 ) -> Union[ReadOnlyColumnCollection[_COLKEY, _COL_co], _COL_co]:
1945 try:
1946 if isinstance(key, (tuple, slice)):
1947 if isinstance(key, slice):
1948 cols = (
1949 (sub_key, col)
1950 for (sub_key, col, _) in self._collection[key]
1951 )
1952 else:
1953 cols = (self._index[sub_key] for sub_key in key)
1955 return WriteableColumnCollection(cols).as_readonly()
1956 else:
1957 return self._index[key][1]
1958 except KeyError as err:
1959 if isinstance(err.args[0], int):
1960 raise IndexError(err.args[0]) from err
1961 else:
1962 raise
1964 def __getattr__(self, key: str) -> _COL_co:
1965 try:
1966 return self._index[key][1]
1967 except KeyError as err:
1968 raise AttributeError(key) from err
1970 def __contains__(self, key: str) -> bool:
1971 if key not in self._index:
1972 if not isinstance(key, str):
1973 raise exc.ArgumentError(
1974 "__contains__ requires a string argument"
1975 )
1976 return False
1977 else:
1978 return True
1980 def compare(self, other: ColumnCollection[_COLKEY, _COL_co]) -> bool:
1981 """Compare this :class:`_expression.ColumnCollection` to another
1982 based on the names of the keys"""
1984 for l, r in zip_longest(self, other):
1985 if l is not r:
1986 return False
1987 else:
1988 return True
1990 def __eq__(self, other: Any) -> bool:
1991 return self.compare(other)
1993 @overload
1994 def get(self, key: str, default: None = None) -> Optional[_COL_co]: ...
1996 @overload
1997 def get(self, key: str, default: _COL) -> Union[_COL_co, _COL]: ...
1999 def get(
2000 self, key: str, default: Optional[_COL] = None
2001 ) -> Optional[Union[_COL_co, _COL]]:
2002 """Get a :class:`_sql.ColumnClause` or :class:`_schema.Column` object
2003 based on a string key name from this
2004 :class:`_expression.ColumnCollection`."""
2006 if key in self._index:
2007 return self._index[key][1]
2008 else:
2009 return default
2011 def __str__(self) -> str:
2012 return "%s(%s)" % (
2013 self.__class__.__name__,
2014 ", ".join(str(c) for c in self),
2015 )
2017 # https://github.com/python/mypy/issues/4266
2018 __hash__: Optional[int] = None # type: ignore[assignment]
2020 def contains_column(self, col: ColumnElement[Any]) -> bool:
2021 """Checks if a column object exists in this collection"""
2022 if col not in self._colset:
2023 if isinstance(col, str):
2024 raise exc.ArgumentError(
2025 "contains_column cannot be used with string arguments. "
2026 "Use ``col_name in table.c`` instead."
2027 )
2028 return False
2029 else:
2030 return True
2032 def _as_readonly(self) -> ReadOnlyColumnCollection[_COLKEY, _COL_co]:
2033 raise NotImplementedError()
2035 def corresponding_column(
2036 self, column: _COL, require_embedded: bool = False
2037 ) -> Optional[Union[_COL, _COL_co]]:
2038 """Given a :class:`_expression.ColumnElement`, return the exported
2039 :class:`_expression.ColumnElement` object from this
2040 :class:`_expression.ColumnCollection`
2041 which corresponds to that original :class:`_expression.ColumnElement`
2042 via a common
2043 ancestor column.
2045 :param column: the target :class:`_expression.ColumnElement`
2046 to be matched.
2048 :param require_embedded: only return corresponding columns for
2049 the given :class:`_expression.ColumnElement`, if the given
2050 :class:`_expression.ColumnElement`
2051 is actually present within a sub-element
2052 of this :class:`_expression.Selectable`.
2053 Normally the column will match if
2054 it merely shares a common ancestor with one of the exported
2055 columns of this :class:`_expression.Selectable`.
2057 .. seealso::
2059 :meth:`_expression.Selectable.corresponding_column`
2060 - invokes this method
2061 against the collection returned by
2062 :attr:`_expression.Selectable.exported_columns`.
2064 .. versionchanged:: 1.4 the implementation for ``corresponding_column``
2065 was moved onto the :class:`_expression.ColumnCollection` itself.
2067 """
2068 raise NotImplementedError()
2071class WriteableColumnCollection(ColumnCollection[_COLKEY, _COL_co]):
2072 """A :class:`_sql.ColumnCollection` that allows mutation operations.
2074 This is the writable form of :class:`_sql.ColumnCollection` that
2075 implements methods such as :meth:`.add`, :meth:`.remove`, :meth:`.update`,
2076 and :meth:`.clear`.
2078 This class is used internally for building column collections during
2079 construction of SQL constructs. For schema-level objects that require
2080 deduplication behavior, use :class:`.DedupeColumnCollection`.
2082 .. versionadded:: 2.1
2084 """
2086 __slots__ = ()
2088 def __init__(
2089 self, columns: Optional[Iterable[Tuple[_COLKEY, _COL_co]]] = None
2090 ):
2091 object.__setattr__(self, "_colset", set())
2092 object.__setattr__(self, "_index", {})
2093 object.__setattr__(
2094 self, "_proxy_index", collections.defaultdict(util.OrderedSet)
2095 )
2096 object.__setattr__(self, "_collection", [])
2097 if columns:
2098 self._initial_populate(columns)
2100 def _initial_populate(
2101 self, iter_: Iterable[Tuple[_COLKEY, _COL_co]]
2102 ) -> None:
2103 self._populate_separate_keys(iter_)
2105 def _populate_separate_keys(
2106 self, iter_: Iterable[Tuple[_COLKEY, _COL_co]]
2107 ) -> None:
2108 """populate from an iterator of (key, column)"""
2110 self._collection[:] = collection = [
2111 (k, c, _ColumnMetrics(self, c)) for k, c in iter_
2112 ]
2113 self._colset.update(c._deannotate() for _, c, _ in collection)
2114 self._index.update(
2115 {idx: (k, c) for idx, (k, c, _) in enumerate(collection)}
2116 )
2117 self._index.update({k: (k, col) for k, col, _ in reversed(collection)})
2119 def __getstate__(self) -> Dict[str, Any]:
2120 return {
2121 "_collection": [(k, c) for k, c, _ in self._collection],
2122 "_index": self._index,
2123 }
2125 def __setstate__(self, state: Dict[str, Any]) -> None:
2126 object.__setattr__(self, "_index", state["_index"])
2127 object.__setattr__(
2128 self, "_proxy_index", collections.defaultdict(util.OrderedSet)
2129 )
2130 object.__setattr__(
2131 self,
2132 "_collection",
2133 [
2134 (k, c, _ColumnMetrics(self, c))
2135 for (k, c) in state["_collection"]
2136 ],
2137 )
2138 object.__setattr__(
2139 self, "_colset", {col for k, col, _ in self._collection}
2140 )
2142 def add(
2143 self,
2144 column: ColumnElement[Any],
2145 key: Optional[_COLKEY] = None,
2146 ) -> None:
2147 """Add a column to this :class:`_sql.WriteableColumnCollection`.
2149 .. note::
2151 This method is **not normally used by user-facing code**, as the
2152 :class:`_sql.WriteableColumnCollection` is usually part of an
2153 existing object such as a :class:`_schema.Table`. To add a
2154 :class:`_schema.Column` to an existing :class:`_schema.Table`
2155 object, use the :meth:`_schema.Table.append_column` method.
2157 """
2158 colkey: _COLKEY
2160 if key is None:
2161 colkey = column.key # type: ignore[assignment]
2162 else:
2163 colkey = key
2165 l = len(self._collection)
2167 # don't really know how this part is supposed to work w/ the
2168 # covariant thing
2170 _column = cast(_COL_co, column)
2172 self._collection.append(
2173 (colkey, _column, _ColumnMetrics(self, _column))
2174 )
2175 self._colset.add(_column._deannotate())
2177 self._index[l] = (colkey, _column)
2178 if colkey not in self._index:
2179 self._index[colkey] = (colkey, _column)
2181 def _as_readonly(self) -> ReadOnlyColumnCollection[_COLKEY, _COL_co]:
2182 return ReadOnlyColumnCollection(self)
2184 def as_readonly(self) -> ReadOnlyColumnCollection[_COLKEY, _COL_co]:
2185 """Return a "read only" form of this
2186 :class:`_sql.WriteableColumnCollection`."""
2188 return self._as_readonly()
2190 def _init_proxy_index(self) -> None:
2191 """populate the "proxy index", if empty.
2193 proxy index is added in 2.0 to provide more efficient operation
2194 for the corresponding_column() method.
2196 For reasons of both time to construct new .c collections as well as
2197 memory conservation for large numbers of large .c collections, the
2198 proxy_index is only filled if corresponding_column() is called. once
2199 filled it stays that way, and new _ColumnMetrics objects created after
2200 that point will populate it with new data. Note this case would be
2201 unusual, if not nonexistent, as it means a .c collection is being
2202 mutated after corresponding_column() were used, however it is tested in
2203 test/base/test_utils.py.
2205 """
2206 pi = self._proxy_index
2207 if pi:
2208 return
2210 for _, _, metrics in self._collection:
2211 eps = metrics.column._expanded_proxy_set
2213 for eps_col in eps:
2214 pi[eps_col].add(metrics)
2216 def corresponding_column(
2217 self, column: _COL, require_embedded: bool = False
2218 ) -> Optional[Union[_COL, _COL_co]]:
2219 """Given a :class:`_expression.ColumnElement`, return the exported
2220 :class:`_expression.ColumnElement` object from this
2221 :class:`_expression.ColumnCollection`
2222 which corresponds to that original :class:`_expression.ColumnElement`
2223 via a common
2224 ancestor column.
2226 See :meth:`.ColumnCollection.corresponding_column` for parameter
2227 information.
2229 """
2230 # TODO: cython candidate
2232 # don't dig around if the column is locally present
2233 if column in self._colset:
2234 return column
2236 selected_intersection, selected_metrics = None, None
2237 target_set = column.proxy_set
2239 pi = self._proxy_index
2240 if not pi:
2241 self._init_proxy_index()
2243 for current_metrics in (
2244 mm for ts in target_set if ts in pi for mm in pi[ts]
2245 ):
2246 if not require_embedded or current_metrics.embedded(target_set):
2247 if selected_metrics is None:
2248 # no corresponding column yet, pick this one.
2249 selected_metrics = current_metrics
2250 continue
2252 current_intersection = target_set.intersection(
2253 current_metrics.column._expanded_proxy_set
2254 )
2255 if selected_intersection is None:
2256 selected_intersection = target_set.intersection(
2257 selected_metrics.column._expanded_proxy_set
2258 )
2260 if len(current_intersection) > len(selected_intersection):
2261 # 'current' has a larger field of correspondence than
2262 # 'selected'. i.e. selectable.c.a1_x->a1.c.x->table.c.x
2263 # matches a1.c.x->table.c.x better than
2264 # selectable.c.x->table.c.x does.
2266 selected_metrics = current_metrics
2267 selected_intersection = current_intersection
2268 elif current_intersection == selected_intersection:
2269 # they have the same field of correspondence. see
2270 # which proxy_set has fewer columns in it, which
2271 # indicates a closer relationship with the root
2272 # column. Also take into account the "weight"
2273 # attribute which CompoundSelect() uses to give
2274 # higher precedence to columns based on vertical
2275 # position in the compound statement, and discard
2276 # columns that have no reference to the target
2277 # column (also occurs with CompoundSelect)
2279 selected_col_distance = sum(
2280 [
2281 sc._annotations.get("weight", 1)
2282 for sc in (
2283 selected_metrics.column._uncached_proxy_list()
2284 )
2285 if sc.shares_lineage(column)
2286 ],
2287 )
2288 current_col_distance = sum(
2289 [
2290 sc._annotations.get("weight", 1)
2291 for sc in (
2292 current_metrics.column._uncached_proxy_list()
2293 )
2294 if sc.shares_lineage(column)
2295 ],
2296 )
2297 if current_col_distance < selected_col_distance:
2298 selected_metrics = current_metrics
2299 selected_intersection = current_intersection
2301 return selected_metrics.column if selected_metrics else None
2304_NAMEDCOL = TypeVar("_NAMEDCOL", bound="NamedColumn[Any]")
2307class DedupeColumnCollection(WriteableColumnCollection[str, _NAMEDCOL]):
2308 """A :class:`_expression.ColumnCollection`
2309 that maintains deduplicating behavior.
2311 This is useful by schema level objects such as :class:`_schema.Table` and
2312 :class:`.PrimaryKeyConstraint`. The collection includes more
2313 sophisticated mutator methods as well to suit schema objects which
2314 require mutable column collections.
2316 .. versionadded:: 1.4
2318 """
2320 def add( # type: ignore[override]
2321 self,
2322 column: _NAMEDCOL,
2323 key: Optional[str] = None,
2324 *,
2325 index: Optional[int] = None,
2326 ) -> None:
2327 if key is not None and column.key != key:
2328 raise exc.ArgumentError(
2329 "DedupeColumnCollection requires columns be under "
2330 "the same key as their .key"
2331 )
2332 key = column.key
2334 if key is None:
2335 raise exc.ArgumentError(
2336 "Can't add unnamed column to column collection"
2337 )
2339 if key in self._index:
2340 existing = self._index[key][1]
2342 if existing is column:
2343 return
2345 self.replace(column, index=index)
2347 # pop out memoized proxy_set as this
2348 # operation may very well be occurring
2349 # in a _make_proxy operation
2350 util.memoized_property.reset(column, "proxy_set")
2351 else:
2352 self._append_new_column(key, column, index=index)
2354 def _append_new_column(
2355 self, key: str, named_column: _NAMEDCOL, *, index: Optional[int] = None
2356 ) -> None:
2357 collection_length = len(self._collection)
2359 if index is None:
2360 l = collection_length
2361 else:
2362 if index < 0:
2363 index = max(0, collection_length + index)
2364 l = index
2366 if index is None:
2367 self._collection.append(
2368 (key, named_column, _ColumnMetrics(self, named_column))
2369 )
2370 else:
2371 self._collection.insert(
2372 index, (key, named_column, _ColumnMetrics(self, named_column))
2373 )
2375 self._colset.add(named_column._deannotate())
2377 if index is not None:
2378 for idx in reversed(range(index, collection_length)):
2379 self._index[idx + 1] = self._index[idx]
2381 self._index[l] = (key, named_column)
2382 self._index[key] = (key, named_column)
2384 def _populate_separate_keys(
2385 self, iter_: Iterable[Tuple[str, _NAMEDCOL]]
2386 ) -> None:
2387 """populate from an iterator of (key, column)"""
2388 cols = list(iter_)
2390 replace_col = []
2391 for k, col in cols:
2392 if col.key != k:
2393 raise exc.ArgumentError(
2394 "DedupeColumnCollection requires columns be under "
2395 "the same key as their .key"
2396 )
2397 if col.name in self._index and col.key != col.name:
2398 replace_col.append(col)
2399 elif col.key in self._index:
2400 replace_col.append(col)
2401 else:
2402 self._index[k] = (k, col)
2403 self._collection.append((k, col, _ColumnMetrics(self, col)))
2404 self._colset.update(c._deannotate() for (k, c, _) in self._collection)
2406 self._index.update(
2407 (idx, (k, c)) for idx, (k, c, _) in enumerate(self._collection)
2408 )
2409 for col in replace_col:
2410 self.replace(col)
2412 def extend(self, iter_: Iterable[_NAMEDCOL]) -> None:
2413 self._populate_separate_keys((col.key, col) for col in iter_)
2415 def remove(self, column: _NAMEDCOL) -> None:
2416 if column not in self._colset:
2417 raise ValueError(
2418 "Can't remove column %r; column is not in this collection"
2419 % column
2420 )
2421 del self._index[column.key]
2422 self._colset.remove(column)
2423 self._collection[:] = [
2424 (k, c, metrics)
2425 for (k, c, metrics) in self._collection
2426 if c is not column
2427 ]
2428 for metrics in self._proxy_index.get(column, ()):
2429 metrics.dispose(self)
2431 self._index.update(
2432 {idx: (k, col) for idx, (k, col, _) in enumerate(self._collection)}
2433 )
2434 # delete higher index
2435 del self._index[len(self._collection)]
2437 def replace(
2438 self,
2439 column: _NAMEDCOL,
2440 *,
2441 extra_remove: Optional[Iterable[_NAMEDCOL]] = None,
2442 index: Optional[int] = None,
2443 ) -> None:
2444 """add the given column to this collection, removing unaliased
2445 versions of this column as well as existing columns with the
2446 same key.
2448 e.g.::
2450 t = Table("sometable", metadata, Column("col1", Integer))
2451 t.columns.replace(Column("col1", Integer, key="columnone"))
2453 will remove the original 'col1' from the collection, and add
2454 the new column under the name 'columnname'.
2456 Used by schema.Column to override columns during table reflection.
2458 """
2460 if extra_remove:
2461 remove_col = set(extra_remove)
2462 else:
2463 remove_col = set()
2464 # remove up to two columns based on matches of name as well as key
2465 if column.name in self._index and column.key != column.name:
2466 other = self._index[column.name][1]
2467 if other.name == other.key:
2468 remove_col.add(other)
2470 if column.key in self._index:
2471 remove_col.add(self._index[column.key][1])
2473 if not remove_col:
2474 self._append_new_column(column.key, column, index=index)
2475 return
2476 new_cols: List[Tuple[str, _NAMEDCOL, _ColumnMetrics[_NAMEDCOL]]] = []
2477 replace_index = None
2479 for idx, (k, col, metrics) in enumerate(self._collection):
2480 if col in remove_col:
2481 if replace_index is None:
2482 replace_index = idx
2483 new_cols.append(
2484 (column.key, column, _ColumnMetrics(self, column))
2485 )
2486 else:
2487 new_cols.append((k, col, metrics))
2489 if remove_col:
2490 self._colset.difference_update(remove_col)
2492 for rc in remove_col:
2493 for metrics in self._proxy_index.get(rc, ()):
2494 metrics.dispose(self)
2496 if replace_index is None:
2497 if index is not None:
2498 new_cols.insert(
2499 index, (column.key, column, _ColumnMetrics(self, column))
2500 )
2502 else:
2503 new_cols.append(
2504 (column.key, column, _ColumnMetrics(self, column))
2505 )
2506 elif index is not None:
2507 to_move = new_cols[replace_index]
2508 effective_positive_index = (
2509 index if index >= 0 else max(0, len(new_cols) + index)
2510 )
2511 new_cols.insert(index, to_move)
2512 if replace_index > effective_positive_index:
2513 del new_cols[replace_index + 1]
2514 else:
2515 del new_cols[replace_index]
2517 self._colset.add(column._deannotate())
2518 self._collection[:] = new_cols
2520 self._index.clear()
2522 self._index.update(
2523 {idx: (k, col) for idx, (k, col, _) in enumerate(self._collection)}
2524 )
2525 self._index.update({k: (k, col) for (k, col, _) in self._collection})
2528class ReadOnlyColumnCollection(
2529 util.ReadOnlyContainer, ColumnCollection[_COLKEY, _COL_co]
2530):
2531 __slots__ = ("_parent",)
2533 _parent: WriteableColumnCollection[_COLKEY, _COL_co]
2535 def __init__(
2536 self, collection: WriteableColumnCollection[_COLKEY, _COL_co]
2537 ):
2538 object.__setattr__(self, "_parent", collection)
2539 object.__setattr__(self, "_index", collection._index)
2540 object.__setattr__(self, "_collection", collection._collection)
2541 object.__setattr__(self, "_colset", collection._colset)
2542 object.__setattr__(self, "_proxy_index", collection._proxy_index)
2544 def _as_readonly(self) -> ReadOnlyColumnCollection[_COLKEY, _COL_co]:
2545 return self
2547 def __getstate__(self) -> Dict[str, ColumnCollection[_COLKEY, _COL_co]]:
2548 return {"_parent": self._parent}
2550 def __setstate__(self, state: Dict[str, Any]) -> None:
2551 parent = state["_parent"]
2552 self.__init__(parent) # type: ignore[misc]
2554 def corresponding_column(
2555 self, column: _COL, require_embedded: bool = False
2556 ) -> Optional[Union[_COL, _COL_co]]:
2557 """Given a :class:`_expression.ColumnElement`, return the exported
2558 :class:`_expression.ColumnElement` object from this
2559 :class:`_expression.ColumnCollection`
2560 which corresponds to that original :class:`_expression.ColumnElement`
2561 via a common
2562 ancestor column.
2564 See :meth:`.ColumnCollection.corresponding_column` for parameter
2565 information.
2567 """
2568 return self._parent.corresponding_column(column, require_embedded)
2571class ColumnSet(util.OrderedSet["ColumnClause[Any]"]):
2572 def contains_column(self, col: ColumnClause[Any]) -> bool:
2573 return col in self
2575 def extend(self, cols: Iterable[Any]) -> None:
2576 for col in cols:
2577 self.add(col)
2579 def __eq__(self, other):
2580 l = []
2581 for c in other:
2582 for local in self:
2583 if c.shares_lineage(local):
2584 l.append(c == local)
2585 return elements.and_(*l)
2587 def __hash__(self) -> int: # type: ignore[override]
2588 return hash(tuple(x for x in self))
2591def _entity_namespace(
2592 entity: Union[_HasEntityNamespace, ExternallyTraversible],
2593) -> _EntityNamespace:
2594 """Return the nearest .entity_namespace for the given entity.
2596 If not immediately available, does an iterate to find a sub-element
2597 that has one, if any.
2599 """
2600 try:
2601 return cast(_HasEntityNamespace, entity).entity_namespace
2602 except AttributeError:
2603 for elem in visitors.iterate(cast(ExternallyTraversible, entity)):
2604 if _is_has_entity_namespace(elem):
2605 return elem.entity_namespace
2606 else:
2607 raise
2610@overload
2611def _entity_namespace_key(
2612 entity: Union[_HasEntityNamespace, ExternallyTraversible],
2613 key: str,
2614) -> SQLCoreOperations[Any]: ...
2617@overload
2618def _entity_namespace_key(
2619 entity: Union[_HasEntityNamespace, ExternallyTraversible],
2620 key: str,
2621 default: _NoArg,
2622) -> SQLCoreOperations[Any]: ...
2625@overload
2626def _entity_namespace_key(
2627 entity: Union[_HasEntityNamespace, ExternallyTraversible],
2628 key: str,
2629 default: _T,
2630) -> Union[SQLCoreOperations[Any], _T]: ...
2633def _entity_namespace_key(
2634 entity: Union[_HasEntityNamespace, ExternallyTraversible],
2635 key: str,
2636 default: Union[SQLCoreOperations[Any], _T, _NoArg] = NO_ARG,
2637) -> Union[SQLCoreOperations[Any], _T]:
2638 """Return an entry from an entity_namespace.
2641 Raises :class:`_exc.InvalidRequestError` rather than attribute error
2642 on not found.
2644 """
2646 try:
2647 ns = _entity_namespace(entity)
2648 if default is not NO_ARG:
2649 return getattr(ns, key, default)
2650 else:
2651 return getattr(ns, key) # type: ignore[no-any-return]
2652 except AttributeError as err:
2653 raise exc.InvalidRequestError(
2654 'Entity namespace for "%s" has no property "%s"' % (entity, key)
2655 ) from err
2658def _entity_namespace_key_search_all(
2659 entities: Collection[Any],
2660 key: str,
2661) -> SQLCoreOperations[Any]:
2662 """Search multiple entities for a key, raise if ambiguous or not found.
2664 This is used by filter_by() to search across all FROM clause entities
2665 when a single entity doesn't have the requested attribute.
2667 .. versionadded:: 2.1
2669 Raises:
2670 AmbiguousColumnError: If key exists in multiple entities
2671 InvalidRequestError: If key doesn't exist in any entity
2672 """
2674 match_: SQLCoreOperations[Any] | None = None
2676 for entity in entities:
2677 ns = _entity_namespace(entity)
2678 # Check if the attribute exists
2679 if hasattr(ns, key):
2680 if match_ is not None:
2681 entity_desc = ", ".join(str(e) for e in list(entities)[:3])
2682 if len(entities) > 3:
2683 entity_desc += f", ... ({len(entities)} total)"
2684 raise exc.AmbiguousColumnError(
2685 f'Attribute name "{key}" is ambiguous; it exists in '
2686 f"multiple FROM clause entities ({entity_desc}). "
2687 f"Use filter() with explicit column references instead "
2688 f"of filter_by()."
2689 )
2690 match_ = getattr(ns, key)
2692 if match_ is None:
2693 # No entity has this attribute
2694 entity_desc = ", ".join(str(e) for e in list(entities)[:3])
2695 if len(entities) > 3:
2696 entity_desc += f", ... ({len(entities)} total)"
2697 raise exc.InvalidRequestError(
2698 f'None of the FROM clause entities have a property "{key}". '
2699 f"Searched entities: {entity_desc}"
2700 )
2702 return match_