codekingpro/portable-devtools
115k
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
8
9"""Foundational utilities common to many sql modules."""
10
11
12from __future__ import annotations
13
14import collections
15from enum import Enum
16import itertools
17from itertools import zip_longest
18import operator
19import re
20from typing import Any
21from typing import Callable
22from typing import cast
23from typing import Dict
24from typing import FrozenSet
25from typing import Generator
26from typing import Generic
27from typing import Iterable
28from typing import Iterator
29from typing import List
30from typing import Mapping
31from typing import MutableMapping
32from typing import NamedTuple
33from typing import NoReturn
34from typing import Optional
35from typing import overload
36from typing import Sequence
37from typing import Set
38from typing import Tuple
39from typing import Type
40from typing import TYPE_CHECKING
41from typing import TypeVar
42from typing import Union
43
44from . import roles
45from . import visitors
46from .cache_key import HasCacheKey # noqa
47from .cache_key import MemoizedHasCacheKey # noqa
48from .traversals import HasCopyInternals # noqa
49from .visitors import ClauseVisitor
50from .visitors import ExtendedInternalTraversal
51from .visitors import ExternallyTraversible
52from .visitors import InternalTraversal
53from .. import event
54from .. import exc
55from .. import util
56from ..util import HasMemoized as HasMemoized
57from ..util import hybridmethod
58from ..util import typing as compat_typing
59from ..util import warn_deprecated
60from ..util.typing import Final
61from ..util.typing import Protocol
62from ..util.typing import Self
63from ..util.typing import TypeGuard
64
65if TYPE_CHECKING:
66 from . import coercions
67 from . import elements
68 from . import type_api
69 from ._orm_types import DMLStrategyArgument
70 from ._orm_types import SynchronizeSessionArgument
71 from ._typing import _CLE
72 from .cache_key import CacheKey
73 from .compiler import SQLCompiler
74 from .elements import BindParameter
75 from .elements import ClauseList
76 from .elements import ColumnClause # noqa
77 from .elements import ColumnElement
78 from .elements import NamedColumn
79 from .elements import SQLCoreOperations
80 from .elements import TextClause
81 from .schema import Column
82 from .schema import DefaultGenerator
83 from .selectable import _JoinTargetElement
84 from .selectable import _SelectIterable
85 from .selectable import FromClause
86 from .visitors import anon_map
87 from ..engine import Connection
88 from ..engine import CursorResult
89 from ..engine.interfaces import _CoreMultiExecuteParams
90 from ..engine.interfaces import _ExecuteOptions
91 from ..engine.interfaces import _ImmutableExecuteOptions
92 from ..engine.interfaces import CacheStats
93 from ..engine.interfaces import Compiled
94 from ..engine.interfaces import CompiledCacheType
95 from ..engine.interfaces import CoreExecuteOptionsParameter
96 from ..engine.interfaces import Dialect
97 from ..engine.interfaces import IsolationLevel
98 from ..engine.interfaces import SchemaTranslateMapType
99 from ..event import dispatcher
100
101if not TYPE_CHECKING:
102 coercions = None # noqa
103 elements = None # noqa
104 type_api = None # noqa
105
106
107class _NoArg(Enum):
108 NO_ARG = 0
109
110 def __repr__(self):
111 return f"_NoArg.{self.name}"
112
113
114NO_ARG: Final = _NoArg.NO_ARG
115
116
117class _NoneName(Enum):
118 NONE_NAME = 0
119 """indicate a 'deferred' name that was ultimately the value None."""
120
121
122_NONE_NAME: Final = _NoneName.NONE_NAME
123
124_T = TypeVar("_T", bound=Any)
125
126_Fn = TypeVar("_Fn", bound=Callable[..., Any])
127
128_AmbiguousTableNameMap = MutableMapping[str, str]
129
130
131class _DefaultDescriptionTuple(NamedTuple):
132 arg: Any
133 is_scalar: Optional[bool]
134 is_callable: Optional[bool]
135 is_sentinel: Optional[bool]
136
137 @classmethod
138 def _from_column_default(
139 cls, default: Optional[DefaultGenerator]
140 ) -> _DefaultDescriptionTuple:
141 return (
142 _DefaultDescriptionTuple(
143 default.arg, # type: ignore
144 default.is_scalar,
145 default.is_callable,
146 default.is_sentinel,
147 )
148 if default
149 and (
150 default.has_arg
151 or (not default.for_update and default.is_sentinel)
152 )
153 else _DefaultDescriptionTuple(None, None, None, None)
154 )
155
156
157_never_select_column: operator.attrgetter[Any] = operator.attrgetter(
158 "_omit_from_statements"
159)
160
161
162class _EntityNamespace(Protocol):
163 def __getattr__(self, key: str) -> SQLCoreOperations[Any]: ...
164
165
166class _HasEntityNamespace(Protocol):
167 @util.ro_non_memoized_property
168 def entity_namespace(self) -> _EntityNamespace: ...
169
170
171def _is_has_entity_namespace(element: Any) -> TypeGuard[_HasEntityNamespace]:
172 return hasattr(element, "entity_namespace")
173
174
175# Remove when https://github.com/python/mypy/issues/14640 will be fixed
176_Self = TypeVar("_Self", bound=Any)
177
178
179class Immutable:
180 """mark a ClauseElement as 'immutable' when expressions are cloned.
181
182 "immutable" objects refers to the "mutability" of an object in the
183 context of SQL DQL and DML generation. Such as, in DQL, one can
184 compose a SELECT or subquery of varied forms, but one cannot modify
185 the structure of a specific table or column within DQL.
186 :class:`.Immutable` is mostly intended to follow this concept, and as
187 such the primary "immutable" objects are :class:`.ColumnClause`,
188 :class:`.Column`, :class:`.TableClause`, :class:`.Table`.
189
190 """
191
192 __slots__ = ()
193
194 _is_immutable: bool = True
195
196 def unique_params(self, *optionaldict: Any, **kwargs: Any) -> NoReturn:
197 raise NotImplementedError("Immutable objects do not support copying")
198
199 def params(self, *optionaldict: Any, **kwargs: Any) -> NoReturn:
200 raise NotImplementedError("Immutable objects do not support copying")
201
202 def _clone(self: _Self, **kw: Any) -> _Self:
203 return self
204
205 def _copy_internals(
206 self, *, omit_attrs: Iterable[str] = (), **kw: Any
207 ) -> None:
208 pass
209
210
211class SingletonConstant(Immutable):
212 """Represent SQL constants like NULL, TRUE, FALSE"""
213
214 _is_singleton_constant: bool = True
215
216 _singleton: SingletonConstant
217
218 def __new__(cls: _T, *arg: Any, **kw: Any) -> _T:
219 return cast(_T, cls._singleton)
220
221 @util.non_memoized_property
222 def proxy_set(self) -> FrozenSet[ColumnElement[Any]]:
223 raise NotImplementedError()
224
225 @classmethod
226 def _create_singleton(cls) -> None:
227 obj = object.__new__(cls)
228 obj.__init__() # type: ignore
229
230 # for a long time this was an empty frozenset, meaning
231 # a SingletonConstant would never be a "corresponding column" in
232 # a statement. This referred to #6259. However, in #7154 we see
233 # that we do in fact need "correspondence" to work when matching cols
234 # in result sets, so the non-correspondence was moved to a more
235 # specific level when we are actually adapting expressions for SQL
236 # render only.
237 obj.proxy_set = frozenset([obj])
238 cls._singleton = obj
239
240
241def _from_objects(
242 *elements: Union[
243 ColumnElement[Any], FromClause, TextClause, _JoinTargetElement
244 ]
245) -> Iterator[FromClause]:
246 return itertools.chain.from_iterable(
247 [element._from_objects for element in elements]
248 )
249
250
251def _select_iterables(
252 elements: Iterable[roles.ColumnsClauseRole],
253) -> _SelectIterable:
254 """expand tables into individual columns in the
255 given list of column expressions.
256
257 """
258 return itertools.chain.from_iterable(
259 [c._select_iterable for c in elements]
260 )
261
262
263_SelfGenerativeType = TypeVar("_SelfGenerativeType", bound="_GenerativeType")
264
265
266class _GenerativeType(compat_typing.Protocol):
267 def _generate(self) -> Self: ...
268
269
270def _generative(fn: _Fn) -> _Fn:
271 """non-caching _generative() decorator.
272
273 This is basically the legacy decorator that copies the object and
274 runs a method on the new copy.
275
276 """
277
278 @util.decorator
279 def _generative(
280 fn: _Fn, self: _SelfGenerativeType, *args: Any, **kw: Any
281 ) -> _SelfGenerativeType:
282 """Mark a method as generative."""
283
284 self = self._generate()
285 x = fn(self, *args, **kw)
286 assert x is self, "generative methods must return self"
287 return self
288
289 decorated = _generative(fn)
290 decorated.non_generative = fn # type: ignore
291 return decorated
292
293
294def _exclusive_against(*names: str, **kw: Any) -> Callable[[_Fn], _Fn]:
295 msgs: Dict[str, str] = kw.pop("msgs", {})
296
297 defaults: Dict[str, str] = kw.pop("defaults", {})
298
299 getters: List[Tuple[str, operator.attrgetter[Any], Optional[str]]] = [
300 (name, operator.attrgetter(name), defaults.get(name, None))
301 for name in names
302 ]
303
304 @util.decorator
305 def check(fn: _Fn, *args: Any, **kw: Any) -> Any:
306 # make pylance happy by not including "self" in the argument
307 # list
308 self = args[0]
309 args = args[1:]
310 for name, getter, default_ in getters:
311 if getter(self) is not default_:
312 msg = msgs.get(
313 name,
314 "Method %s() has already been invoked on this %s construct"
315 % (fn.__name__, self.__class__),
316 )
317 raise exc.InvalidRequestError(msg)
318 return fn(self, *args, **kw)
319
320 return check
321
322
323def _clone(element, **kw):
324 return element._clone(**kw)
325
326
327def _expand_cloned(
328 elements: Iterable[_CLE],
329) -> Iterable[_CLE]:
330 """expand the given set of ClauseElements to be the set of all 'cloned'
331 predecessors.
332
333 """
334 # TODO: cython candidate
335 return itertools.chain(*[x._cloned_set for x in elements])
336
337
338def _de_clone(
339 elements: Iterable[_CLE],
340) -> Iterable[_CLE]:
341 for x in elements:
342 while x._is_clone_of is not None:
343 x = x._is_clone_of
344 yield x
345
346
347def _cloned_intersection(a: Iterable[_CLE], b: Iterable[_CLE]) -> Set[_CLE]:
348 """return the intersection of sets a and b, counting
349 any overlap between 'cloned' predecessors.
350
351 The returned set is in terms of the entities present within 'a'.
352
353 """
354 all_overlap: Set[_CLE] = set(_expand_cloned(a)).intersection(
355 _expand_cloned(b)
356 )
357 return {elem for elem in a if all_overlap.intersection(elem._cloned_set)}
358
359
360def _cloned_difference(a: Iterable[_CLE], b: Iterable[_CLE]) -> Set[_CLE]:
361 all_overlap: Set[_CLE] = set(_expand_cloned(a)).intersection(
362 _expand_cloned(b)
363 )
364 return {
365 elem for elem in a if not all_overlap.intersection(elem._cloned_set)
366 }
367
368
369class _DialectArgView(MutableMapping[str, Any]):
370 """A dictionary view of dialect-level arguments in the form
371 <dialectname>_<argument_name>.
372
373 """
374
375 __slots__ = ("obj",)
376
377 def __init__(self, obj: DialectKWArgs) -> None:
378 self.obj = obj
379
380 def _key(self, key: str) -> Tuple[str, str]:
381 try:
382 dialect, value_key = key.split("_", 1)
383 except ValueError as err:
384 raise KeyError(key) from err
385 else:
386 return dialect, value_key
387
388 def __getitem__(self, key: str) -> Any:
389 dialect, value_key = self._key(key)
390
391 try:
392 opt = self.obj.dialect_options[dialect]
393 except exc.NoSuchModuleError as err:
394 raise KeyError(key) from err
395 else:
396 return opt[value_key]
397
398 def __setitem__(self, key: str, value: Any) -> None:
399 try:
400 dialect, value_key = self._key(key)
401 except KeyError as err:
402 raise exc.ArgumentError(
403 "Keys must be of the form <dialectname>_<argname>"
404 ) from err
405 else:
406 self.obj.dialect_options[dialect][value_key] = value
407
408 def __delitem__(self, key: str) -> None:
409 dialect, value_key = self._key(key)
410 del self.obj.dialect_options[dialect][value_key]
411
412 def __len__(self) -> int:
413 return sum(
414 len(args._non_defaults)
415 for args in self.obj.dialect_options.values()
416 )
417
418 def __iter__(self) -> Generator[str, None, None]:
419 return (
420 "%s_%s" % (dialect_name, value_name)
421 for dialect_name in self.obj.dialect_options
422 for value_name in self.obj.dialect_options[
423 dialect_name
424 ]._non_defaults
425 )
426
427
428class _DialectArgDict(MutableMapping[str, Any]):
429 """A dictionary view of dialect-level arguments for a specific
430 dialect.
431
432 Maintains a separate collection of user-specified arguments
433 and dialect-specified default arguments.
434
435 """
436
437 def __init__(self) -> None:
438 self._non_defaults: Dict[str, Any] = {}
439 self._defaults: Dict[str, Any] = {}
440
441 def __len__(self) -> int:
442 return len(set(self._non_defaults).union(self._defaults))
443
444 def __iter__(self) -> Iterator[str]:
445 return iter(set(self._non_defaults).union(self._defaults))
446
447 def __getitem__(self, key: str) -> Any:
448 if key in self._non_defaults:
449 return self._non_defaults[key]
450 else:
451 return self._defaults[key]
452
453 def __setitem__(self, key: str, value: Any) -> None:
454 self._non_defaults[key] = value
455
456 def __delitem__(self, key: str) -> None:
457 del self._non_defaults[key]
458
459
460@util.preload_module("sqlalchemy.dialects")
461def _kw_reg_for_dialect(dialect_name: str) -> Optional[Dict[Any, Any]]:
462 dialect_cls = util.preloaded.dialects.registry.load(dialect_name)
463 if dialect_cls.construct_arguments is None:
464 return None
465 return dict(dialect_cls.construct_arguments)
466
467
468class DialectKWArgs:
469 """Establish the ability for a class to have dialect-specific arguments
470 with defaults and constructor validation.
471
472 The :class:`.DialectKWArgs` interacts with the
473 :attr:`.DefaultDialect.construct_arguments` present on a dialect.
474
475 .. seealso::
476
477 :attr:`.DefaultDialect.construct_arguments`
478
479 """
480
481 __slots__ = ()
482
483 _dialect_kwargs_traverse_internals: List[Tuple[str, Any]] = [
484 ("dialect_options", InternalTraversal.dp_dialect_options)
485 ]
486
487 def get_dialect_option(
488 self,
489 dialect: Dialect,
490 argument_name: str,
491 *,
492 else_: Any = None,
493 deprecated_fallback: Optional[str] = None,
494 ) -> Any:
495 r"""Return the value of a dialect-specific option, or *else_* if
496 this dialect does not register the given argument.
497
498 This is useful for DDL compilers that may be inherited by
499 third-party dialects whose ``construct_arguments`` do not
500 include the same set of keys as the parent dialect.
501
502 :param dialect: The dialect for which to retrieve the option.
503 :param argument_name: The name of the argument to retrieve.
504 :param else\_: The value to return if the argument is not present.
505 :param deprecated_fallback: Optional dialect name to fall back to
506 if the argument is not present for the current dialect. If the
507 argument is present for the fallback dialect but not the current
508 dialect, a deprecation warning will be emitted.
509
510 """
511
512 registry = DialectKWArgs._kw_registry[dialect.name]
513 if registry is None:
514 return else_
515
516 if argument_name in registry.get(self.__class__, {}):
517 if (
518 deprecated_fallback is None
519 or dialect.name == deprecated_fallback
520 ):
521 return self.dialect_options[dialect.name][argument_name]
522
523 # deprecated_fallback is present; need to look in two places
524
525 # Current dialect has this option registered.
526 # Check if user explicitly set it.
527 if (
528 dialect.name in self.dialect_options
529 and argument_name
530 in self.dialect_options[dialect.name]._non_defaults
531 ):
532 # User explicitly set this dialect's option - use it
533 return self.dialect_options[dialect.name][argument_name]
534
535 # User didn't set current dialect's option.
536 # Check for deprecated fallback.
537 elif (
538 deprecated_fallback in self.dialect_options
539 and argument_name
540 in self.dialect_options[deprecated_fallback]._non_defaults
541 ):
542 # User set fallback option but not current dialect's option
543 warn_deprecated(
544 f"Using '{deprecated_fallback}_{argument_name}' "
545 f"with the '{dialect.name}' dialect is deprecated; "
546 f"please additionally specify "
547 f"'{dialect.name}_{argument_name}'.",
548 version="2.1",
549 )
550 return self.dialect_options[deprecated_fallback][argument_name]
551
552 # Return default value
553 return self.dialect_options[dialect.name][argument_name]
554 else:
555 # Current dialect doesn't have the option registered at all.
556 # Don't warn - if a third-party dialect doesn't support an
557 # option, that's their choice, not a deprecation case.
558 return else_
559
560 @classmethod
561 def argument_for(
562 cls, dialect_name: str, argument_name: str, default: Any
563 ) -> None:
564 """Add a new kind of dialect-specific keyword argument for this class.
565
566 E.g.::
567
568 Index.argument_for("mydialect", "length", None)
569
570 some_index = Index("a", "b", mydialect_length=5)
571
572 The :meth:`.DialectKWArgs.argument_for` method is a per-argument
573 way adding extra arguments to the
574 :attr:`.DefaultDialect.construct_arguments` dictionary. This
575 dictionary provides a list of argument names accepted by various
576 schema-level constructs on behalf of a dialect.
577
578 New dialects should typically specify this dictionary all at once as a
579 data member of the dialect class. The use case for ad-hoc addition of
580 argument names is typically for end-user code that is also using
581 a custom compilation scheme which consumes the additional arguments.
582
583 :param dialect_name: name of a dialect. The dialect must be
584 locatable, else a :class:`.NoSuchModuleError` is raised. The
585 dialect must also include an existing
586 :attr:`.DefaultDialect.construct_arguments` collection, indicating
587 that it participates in the keyword-argument validation and default
588 system, else :class:`.ArgumentError` is raised. If the dialect does
589 not include this collection, then any keyword argument can be
590 specified on behalf of this dialect already. All dialects packaged
591 within SQLAlchemy include this collection, however for third party
592 dialects, support may vary.
593
594 :param argument_name: name of the parameter.
595
596 :param default: default value of the parameter.
597
598 """
599
600 construct_arg_dictionary: Optional[Dict[Any, Any]] = (
601 DialectKWArgs._kw_registry[dialect_name]
602 )
603 if construct_arg_dictionary is None:
604 raise exc.ArgumentError(
605 "Dialect '%s' does have keyword-argument "
606 "validation and defaults enabled configured" % dialect_name
607 )
608 if cls not in construct_arg_dictionary:
609 construct_arg_dictionary[cls] = {}
610 construct_arg_dictionary[cls][argument_name] = default
611
612 @property
613 def dialect_kwargs(self) -> _DialectArgView:
614 """A collection of keyword arguments specified as dialect-specific
615 options to this construct.
616
617 The arguments are present here in their original ``<dialect>_<kwarg>``
618 format. Only arguments that were actually passed are included;
619 unlike the :attr:`.DialectKWArgs.dialect_options` collection, which
620 contains all options known by this dialect including defaults.
621
622 The collection is also writable; keys are accepted of the
623 form ``<dialect>_<kwarg>`` where the value will be assembled
624 into the list of options.
625
626 .. seealso::
627
628 :attr:`.DialectKWArgs.dialect_options` - nested dictionary form
629
630 """
631 return _DialectArgView(self)
632
633 @property
634 def kwargs(self) -> _DialectArgView:
635 """A synonym for :attr:`.DialectKWArgs.dialect_kwargs`."""
636 return self.dialect_kwargs
637
638 _kw_registry: util.PopulateDict[str, Optional[Dict[Any, Any]]] = (
639 util.PopulateDict(_kw_reg_for_dialect)
640 )
641
642 @classmethod
643 def _kw_reg_for_dialect_cls(cls, dialect_name: str) -> _DialectArgDict:
644 construct_arg_dictionary = DialectKWArgs._kw_registry[dialect_name]
645 d = _DialectArgDict()
646
647 if construct_arg_dictionary is None:
648 d._defaults.update({"*": None})
649 else:
650 for cls in reversed(cls.__mro__):
651 if cls in construct_arg_dictionary:
652 d._defaults.update(construct_arg_dictionary[cls])
653 return d
654
655 @util.memoized_property
656 def dialect_options(self) -> util.PopulateDict[str, _DialectArgDict]:
657 """A collection of keyword arguments specified as dialect-specific
658 options to this construct.
659
660 This is a two-level nested registry, keyed to ``<dialect_name>``
661 and ``<argument_name>``. For example, the ``postgresql_where``
662 argument would be locatable as::
663
664 arg = my_object.dialect_options["postgresql"]["where"]
665
666 .. versionadded:: 0.9.2
667
668 .. seealso::
669
670 :attr:`.DialectKWArgs.dialect_kwargs` - flat dictionary form
671
672 """
673
674 return util.PopulateDict(self._kw_reg_for_dialect_cls)
675
676 def _validate_dialect_kwargs(self, kwargs: Dict[str, Any]) -> None:
677 # validate remaining kwargs that they all specify DB prefixes
678
679 if not kwargs:
680 return
681
682 for k in kwargs:
683 m = re.match("^(.+?)_(.+)$", k)
684 if not m:
685 raise TypeError(
686 "Additional arguments should be "
687 "named <dialectname>_<argument>, got '%s'" % k
688 )
689 dialect_name, arg_name = m.group(1, 2)
690
691 try:
692 construct_arg_dictionary = self.dialect_options[dialect_name]
693 except exc.NoSuchModuleError:
694 util.warn(
695 "Can't validate argument %r; can't "
696 "locate any SQLAlchemy dialect named %r"
697 % (k, dialect_name)
698 )
699 self.dialect_options[dialect_name] = d = _DialectArgDict()
700 d._defaults.update({"*": None})
701 d._non_defaults[arg_name] = kwargs[k]
702 else:
703 if (
704 "*" not in construct_arg_dictionary
705 and arg_name not in construct_arg_dictionary
706 ):
707 raise exc.ArgumentError(
708 "Argument %r is not accepted by "
709 "dialect %r on behalf of %r"
710 % (k, dialect_name, self.__class__)
711 )
712 else:
713 construct_arg_dictionary[arg_name] = kwargs[k]
714
715
716class CompileState:
717 """Produces additional object state necessary for a statement to be
718 compiled.
719
720 the :class:`.CompileState` class is at the base of classes that assemble
721 state for a particular statement object that is then used by the
722 compiler. This process is essentially an extension of the process that
723 the SQLCompiler.visit_XYZ() method takes, however there is an emphasis
724 on converting raw user intent into more organized structures rather than
725 producing string output. The top-level :class:`.CompileState` for the
726 statement being executed is also accessible when the execution context
727 works with invoking the statement and collecting results.
728
729 The production of :class:`.CompileState` is specific to the compiler, such
730 as within the :meth:`.SQLCompiler.visit_insert`,
731 :meth:`.SQLCompiler.visit_select` etc. methods. These methods are also
732 responsible for associating the :class:`.CompileState` with the
733 :class:`.SQLCompiler` itself, if the statement is the "toplevel" statement,
734 i.e. the outermost SQL statement that's actually being executed.
735 There can be other :class:`.CompileState` objects that are not the
736 toplevel, such as when a SELECT subquery or CTE-nested
737 INSERT/UPDATE/DELETE is generated.
738
739 .. versionadded:: 1.4
740
741 """
742
743 __slots__ = ("statement", "_ambiguous_table_name_map")
744
745 plugins: Dict[Tuple[str, str], Type[CompileState]] = {}
746
747 _ambiguous_table_name_map: Optional[_AmbiguousTableNameMap]
748
749 @classmethod
750 def create_for_statement(
751 cls, statement: Executable, compiler: SQLCompiler, **kw: Any
752 ) -> CompileState:
753 # factory construction.
754
755 if statement._propagate_attrs:
756 plugin_name = statement._propagate_attrs.get(
757 "compile_state_plugin", "default"
758 )
759 klass = cls.plugins.get(
760 (plugin_name, statement._effective_plugin_target), None
761 )
762 if klass is None:
763 klass = cls.plugins[
764 ("default", statement._effective_plugin_target)
765 ]
766
767 else:
768 klass = cls.plugins[
769 ("default", statement._effective_plugin_target)
770 ]
771
772 if klass is cls:
773 return cls(statement, compiler, **kw)
774 else:
775 return klass.create_for_statement(statement, compiler, **kw)
776
777 def __init__(self, statement, compiler, **kw):
778 self.statement = statement
779
780 @classmethod
781 def get_plugin_class(
782 cls, statement: Executable
783 ) -> Optional[Type[CompileState]]:
784 plugin_name = statement._propagate_attrs.get(
785 "compile_state_plugin", None
786 )
787
788 if plugin_name:
789 key = (plugin_name, statement._effective_plugin_target)
790 if key in cls.plugins:
791 return cls.plugins[key]
792
793 # there's no case where we call upon get_plugin_class() and want
794 # to get None back, there should always be a default. return that
795 # if there was no plugin-specific class (e.g. Insert with "orm"
796 # plugin)
797 try:
798 return cls.plugins[("default", statement._effective_plugin_target)]
799 except KeyError:
800 return None
801
802 @classmethod
803 def _get_plugin_class_for_plugin(
804 cls, statement: Executable, plugin_name: str
805 ) -> Optional[Type[CompileState]]:
806 try:
807 return cls.plugins[
808 (plugin_name, statement._effective_plugin_target)
809 ]
810 except KeyError:
811 return None
812
813 @classmethod
814 def plugin_for(
815 cls, plugin_name: str, visit_name: str
816 ) -> Callable[[_Fn], _Fn]:
817 def decorate(cls_to_decorate):
818 cls.plugins[(plugin_name, visit_name)] = cls_to_decorate
819 return cls_to_decorate
820
821 return decorate
822
823
824class Generative(HasMemoized):
825 """Provide a method-chaining pattern in conjunction with the
826 @_generative decorator."""
827
828 def _generate(self) -> Self:
829 skip = self._memoized_keys
830 cls = self.__class__
831 s = cls.__new__(cls)
832 if skip:
833 # ensure this iteration remains atomic
834 s.__dict__ = {
835 k: v for k, v in self.__dict__.copy().items() if k not in skip
836 }
837 else:
838 s.__dict__ = self.__dict__.copy()
839 return s
840
841
842class InPlaceGenerative(HasMemoized):
843 """Provide a method-chaining pattern in conjunction with the
844 @_generative decorator that mutates in place."""
845
846 __slots__ = ()
847
848 def _generate(self) -> Self:
849 skip = self._memoized_keys
850 # note __dict__ needs to be in __slots__ if this is used
851 for k in skip:
852 self.__dict__.pop(k, None)
853 return self
854
855
856class HasCompileState(Generative):
857 """A class that has a :class:`.CompileState` associated with it."""
858
859 _compile_state_plugin: Optional[Type[CompileState]] = None
860
861 _attributes: util.immutabledict[str, Any] = util.EMPTY_DICT
862
863 _compile_state_factory = CompileState.create_for_statement
864
865
866class _MetaOptions(type):
867 """metaclass for the Options class.
868
869 This metaclass is actually necessary despite the availability of the
870 ``__init_subclass__()`` hook as this type also provides custom class-level
871 behavior for the ``__add__()`` method.
872
873 """
874
875 _cache_attrs: Tuple[str, ...]
876
877 def __add__(self, other):
878 o1 = self()
879
880 if set(other).difference(self._cache_attrs):
881 raise TypeError(
882 "dictionary contains attributes not covered by "
883 "Options class %s: %r"
884 % (self, set(other).difference(self._cache_attrs))
885 )
886
887 o1.__dict__.update(other)
888 return o1
889
890 if TYPE_CHECKING:
891
892 def __getattr__(self, key: str) -> Any: ...
893
894 def __setattr__(self, key: str, value: Any) -> None: ...
895
896 def __delattr__(self, key: str) -> None: ...
897
898
899class Options(metaclass=_MetaOptions):
900 """A cacheable option dictionary with defaults."""
901
902 __slots__ = ()
903
904 _cache_attrs: Tuple[str, ...]
905
906 def __init_subclass__(cls) -> None:
907 dict_ = cls.__dict__
908 cls._cache_attrs = tuple(
909 sorted(
910 d
911 for d in dict_
912 if not d.startswith("__")
913 and d not in ("_cache_key_traversal",)
914 )
915 )
916 super().__init_subclass__()
917
918 def __init__(self, **kw: Any) -> None:
919 self.__dict__.update(kw)
920
921 def __add__(self, other):
922 o1 = self.__class__.__new__(self.__class__)
923 o1.__dict__.update(self.__dict__)
924
925 if set(other).difference(self._cache_attrs):
926 raise TypeError(
927 "dictionary contains attributes not covered by "
928 "Options class %s: %r"
929 % (self, set(other).difference(self._cache_attrs))
930 )
931
932 o1.__dict__.update(other)
933 return o1
934
935 def __eq__(self, other):
936 # TODO: very inefficient. This is used only in test suites
937 # right now.
938 for a, b in zip_longest(self._cache_attrs, other._cache_attrs):
939 if getattr(self, a) != getattr(other, b):
940 return False
941 return True
942
943 def __repr__(self) -> str:
944 # TODO: fairly inefficient, used only in debugging right now.
945
946 return "%s(%s)" % (
947 self.__class__.__name__,
948 ", ".join(
949 "%s=%r" % (k, self.__dict__[k])
950 for k in self._cache_attrs
951 if k in self.__dict__
952 ),
953 )
954
955 @classmethod
956 def isinstance(cls, klass: Type[Any]) -> bool:
957 return issubclass(cls, klass)
958
959 @hybridmethod
960 def add_to_element(self, name: str, value: str) -> Any:
961 return self + {name: getattr(self, name) + value}
962
963 @hybridmethod
964 def _state_dict_inst(self) -> Mapping[str, Any]:
965 return self.__dict__
966
967 _state_dict_const: util.immutabledict[str, Any] = util.EMPTY_DICT
968
969 @_state_dict_inst.classlevel
970 def _state_dict(cls) -> Mapping[str, Any]:
971 return cls._state_dict_const
972
973 @classmethod
974 def safe_merge(cls, other: "Options") -> Any:
975 d = other._state_dict()
976
977 # only support a merge with another object of our class
978 # and which does not have attrs that we don't. otherwise
979 # we risk having state that might not be part of our cache
980 # key strategy
981
982 if (
983 cls is not other.__class__
984 and other._cache_attrs
985 and set(other._cache_attrs).difference(cls._cache_attrs)
986 ):
987 raise TypeError(
988 "other element %r is not empty, is not of type %s, "
989 "and contains attributes not covered here %r"
990 % (
991 other,
992 cls,
993 set(other._cache_attrs).difference(cls._cache_attrs),
994 )
995 )
996 return cls + d
997
998 @classmethod
999 def from_execution_options(
1000 cls,
1001 key: str,
1002 attrs: set[str],
1003 exec_options: Mapping[str, Any],
1004 statement_exec_options: Mapping[str, Any],
1005 ) -> Tuple["Options", Mapping[str, Any]]:
1006 """process Options argument in terms of execution options.
1007
1008
1009 e.g.::
1010
1011 (
1012 load_options,
1013 execution_options,
1014 ) = QueryContext.default_load_options.from_execution_options(
1015 "_sa_orm_load_options",
1016 {"populate_existing", "autoflush", "yield_per"},
1017 execution_options,
1018 statement._execution_options,
1019 )
1020
1021 get back the Options and refresh "_sa_orm_load_options" in the
1022 exec options dict w/ the Options as well
1023
1024 """
1025
1026 # common case is that no options we are looking for are
1027 # in either dictionary, so cancel for that first
1028 check_argnames = attrs.intersection(
1029 set(exec_options).union(statement_exec_options)
1030 )
1031
1032 existing_options = exec_options.get(key, cls)
1033
1034 if check_argnames:
1035 result = {}
1036 for argname in check_argnames:
1037 local = "_" + argname
1038 if argname in exec_options:
1039 result[local] = exec_options[argname]
1040 elif argname in statement_exec_options:
1041 result[local] = statement_exec_options[argname]
1042
1043 new_options = existing_options + result
1044 exec_options = util.immutabledict(exec_options).merge_with(
1045 {key: new_options}
1046 )
1047 return new_options, exec_options
1048
1049 else:
1050 return existing_options, exec_options
1051
1052 if TYPE_CHECKING:
1053
1054 def __getattr__(self, key: str) -> Any: ...
1055
1056 def __setattr__(self, key: str, value: Any) -> None: ...
1057
1058 def __delattr__(self, key: str) -> None: ...
1059
1060
1061class CacheableOptions(Options, HasCacheKey):
1062 __slots__ = ()
1063
1064 @hybridmethod
1065 def _gen_cache_key_inst(
1066 self, anon_map: Any, bindparams: List[BindParameter[Any]]
1067 ) -> Optional[Tuple[Any]]:
1068 return HasCacheKey._gen_cache_key(self, anon_map, bindparams)
1069
1070 @_gen_cache_key_inst.classlevel
1071 def _gen_cache_key(
1072 cls, anon_map: "anon_map", bindparams: List[BindParameter[Any]]
1073 ) -> Tuple[CacheableOptions, Any]:
1074 return (cls, ())
1075
1076 @hybridmethod
1077 def _generate_cache_key(self) -> Optional[CacheKey]:
1078 return HasCacheKey._generate_cache_key_for_object(self)
1079
1080
1081class ExecutableOption(HasCopyInternals):
1082 __slots__ = ()
1083
1084 _annotations: _ImmutableExecuteOptions = util.EMPTY_DICT
1085
1086 __visit_name__: str = "executable_option"
1087
1088 _is_has_cache_key: bool = False
1089
1090 _is_core: bool = True
1091
1092 def _clone(self, **kw):
1093 """Create a shallow copy of this ExecutableOption."""
1094 c = self.__class__.__new__(self.__class__)
1095 c.__dict__ = dict(self.__dict__) # type: ignore
1096 return c
1097
1098
1099class Executable(roles.StatementRole):
1100 """Mark a :class:`_expression.ClauseElement` as supporting execution.
1101
1102 :class:`.Executable` is a superclass for all "statement" types
1103 of objects, including :func:`select`, :func:`delete`, :func:`update`,
1104 :func:`insert`, :func:`text`.
1105
1106 """
1107
1108 supports_execution: bool = True
1109 _execution_options: _ImmutableExecuteOptions = util.EMPTY_DICT
1110 _is_default_generator: bool = False
1111 _with_options: Tuple[ExecutableOption, ...] = ()
1112 _with_context_options: Tuple[
1113 Tuple[Callable[[CompileState], None], Any], ...
1114 ] = ()
1115 _compile_options: Optional[Union[Type[CacheableOptions], CacheableOptions]]
1116
1117 _executable_traverse_internals = [
1118 ("_with_options", InternalTraversal.dp_executable_options),
1119 (
1120 "_with_context_options",
1121 ExtendedInternalTraversal.dp_with_context_options,
1122 ),
1123 ("_propagate_attrs", ExtendedInternalTraversal.dp_propagate_attrs),
1124 ]
1125
1126 is_select: bool = False
1127 is_from_statement: bool = False
1128 is_update: bool = False
1129 is_insert: bool = False
1130 is_text: bool = False
1131 is_delete: bool = False
1132 is_dml: bool = False
1133
1134 if TYPE_CHECKING:
1135 __visit_name__: str
1136
1137 def _compile_w_cache(
1138 self,
1139 dialect: Dialect,
1140 *,
1141 compiled_cache: Optional[CompiledCacheType],
1142 column_keys: List[str],
1143 for_executemany: bool = False,
1144 schema_translate_map: Optional[SchemaTranslateMapType] = None,
1145 **kw: Any,
1146 ) -> Tuple[
1147 Compiled, Optional[Sequence[BindParameter[Any]]], CacheStats
1148 ]: ...
1149
1150 def _execute_on_connection(
1151 self,
1152 connection: Connection,
1153 distilled_params: _CoreMultiExecuteParams,
1154 execution_options: CoreExecuteOptionsParameter,
1155 ) -> CursorResult[Any]: ...
1156
1157 def _execute_on_scalar(
1158 self,
1159 connection: Connection,
1160 distilled_params: _CoreMultiExecuteParams,
1161 execution_options: CoreExecuteOptionsParameter,
1162 ) -> Any: ...
1163
1164 @util.ro_non_memoized_property
1165 def _all_selected_columns(self) -> _SelectIterable:
1166 raise NotImplementedError()
1167
1168 @property
1169 def _effective_plugin_target(self) -> str:
1170 return self.__visit_name__
1171
1172 @_generative
1173 def options(self, *options: ExecutableOption) -> Self:
1174 """Apply options to this statement.
1175
1176 In the general sense, options are any kind of Python object
1177 that can be interpreted by the SQL compiler for the statement.
1178 These options can be consumed by specific dialects or specific kinds
1179 of compilers.
1180
1181 The most commonly known kind of option are the ORM level options
1182 that apply "eager load" and other loading behaviors to an ORM
1183 query. However, options can theoretically be used for many other
1184 purposes.
1185
1186 For background on specific kinds of options for specific kinds of
1187 statements, refer to the documentation for those option objects.
1188
1189 .. versionchanged:: 1.4 - added :meth:`.Executable.options` to
1190 Core statement objects towards the goal of allowing unified
1191 Core / ORM querying capabilities.
1192
1193 .. seealso::
1194
1195 :ref:`loading_columns` - refers to options specific to the usage
1196 of ORM queries
1197
1198 :ref:`relationship_loader_options` - refers to options specific
1199 to the usage of ORM queries
1200