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