codekingpro/portable-devtools
114k
1# orm/session.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
8"""Provides the Session class and related utilities."""
9
10from __future__ import annotations
11
12import contextlib
13from enum import Enum
14import itertools
15import sys
16import typing
17from typing import Any
18from typing import Callable
19from typing import cast
20from typing import Dict
21from typing import Generic
22from typing import Iterable
23from typing import Iterator
24from typing import List
25from typing import NoReturn
26from typing import Optional
27from typing import overload
28from typing import Sequence
29from typing import Set
30from typing import Tuple
31from typing import Type
32from typing import TYPE_CHECKING
33from typing import TypeVar
34from typing import Union
35import weakref
36
37from . import attributes
38from . import bulk_persistence
39from . import context
40from . import descriptor_props
41from . import exc
42from . import identity
43from . import loading
44from . import query
45from . import state as statelib
46from ._typing import _O
47from ._typing import insp_is_mapper
48from ._typing import is_composite_class
49from ._typing import is_orm_option
50from ._typing import is_user_defined_option
51from .base import _class_to_mapper
52from .base import _none_set
53from .base import _state_mapper
54from .base import instance_str
55from .base import LoaderCallableStatus
56from .base import object_mapper
57from .base import object_state
58from .base import PassiveFlag
59from .base import state_str
60from .context import FromStatement
61from .context import ORMCompileState
62from .identity import IdentityMap
63from .query import Query
64from .state import InstanceState
65from .state_changes import _StateChange
66from .state_changes import _StateChangeState
67from .state_changes import _StateChangeStates
68from .unitofwork import UOWTransaction
69from .. import engine
70from .. import exc as sa_exc
71from .. import sql
72from .. import util
73from ..engine import Connection
74from ..engine import Engine
75from ..engine.util import TransactionalContext
76from ..event import dispatcher
77from ..event import EventTarget
78from ..inspection import inspect
79from ..inspection import Inspectable
80from ..sql import coercions
81from ..sql import dml
82from ..sql import roles
83from ..sql import Select
84from ..sql import TableClause
85from ..sql import visitors
86from ..sql.base import _NoArg
87from ..sql.base import CompileState
88from ..sql.schema import Table
89from ..sql.selectable import ForUpdateArg
90from ..sql.selectable import LABEL_STYLE_TABLENAME_PLUS_COL
91from ..util import IdentitySet
92from ..util.typing import Literal
93from ..util.typing import Protocol
94
95if typing.TYPE_CHECKING:
96 from ._typing import _EntityType
97 from ._typing import _IdentityKeyType
98 from ._typing import _InstanceDict
99 from ._typing import OrmExecuteOptionsParameter
100 from .interfaces import ORMOption
101 from .interfaces import UserDefinedOption
102 from .mapper import Mapper
103 from .path_registry import PathRegistry
104 from .query import RowReturningQuery
105 from ..engine import CursorResult
106 from ..engine import Result
107 from ..engine import Row
108 from ..engine import RowMapping
109 from ..engine.base import Transaction
110 from ..engine.base import TwoPhaseTransaction
111 from ..engine.interfaces import _CoreAnyExecuteParams
112 from ..engine.interfaces import _CoreSingleExecuteParams
113 from ..engine.interfaces import _ExecuteOptions
114 from ..engine.interfaces import CoreExecuteOptionsParameter
115 from ..engine.result import ScalarResult
116 from ..event import _InstanceLevelDispatch
117 from ..sql._typing import _ColumnsClauseArgument
118 from ..sql._typing import _InfoType
119 from ..sql._typing import _T0
120 from ..sql._typing import _T1
121 from ..sql._typing import _T2
122 from ..sql._typing import _T3
123 from ..sql._typing import _T4
124 from ..sql._typing import _T5
125 from ..sql._typing import _T6
126 from ..sql._typing import _T7
127 from ..sql._typing import _TypedColumnClauseArgument as _TCCA
128 from ..sql.base import Executable
129 from ..sql.base import ExecutableOption
130 from ..sql.dml import UpdateBase
131 from ..sql.elements import ClauseElement
132 from ..sql.roles import TypedColumnsClauseRole
133 from ..sql.selectable import ForUpdateParameter
134 from ..sql.selectable import TypedReturnsRows
135
136_T = TypeVar("_T", bound=Any)
137
138__all__ = [
139 "Session",
140 "SessionTransaction",
141 "sessionmaker",
142 "ORMExecuteState",
143 "close_all_sessions",
144 "make_transient",
145 "make_transient_to_detached",
146 "object_session",
147]
148
149_sessions: weakref.WeakValueDictionary[int, Session] = (
150 weakref.WeakValueDictionary()
151)
152"""Weak-referencing dictionary of :class:`.Session` objects.
153"""
154
155statelib._sessions = _sessions
156
157_PKIdentityArgument = Union[Any, Tuple[Any, ...]]
158
159_BindArguments = Dict[str, Any]
160
161_EntityBindKey = Union[Type[_O], "Mapper[_O]"]
162_SessionBindKey = Union[Type[Any], "Mapper[Any]", "TableClause", str]
163_SessionBind = Union["Engine", "Connection"]
164
165JoinTransactionMode = Literal[
166 "conditional_savepoint",
167 "rollback_only",
168 "control_fully",
169 "create_savepoint",
170]
171
172
173class _ConnectionCallableProto(Protocol):
174 """a callable that returns a :class:`.Connection` given an instance.
175
176 This callable, when present on a :class:`.Session`, is called only from the
177 ORM's persistence mechanism (i.e. the unit of work flush process) to allow
178 for connection-per-instance schemes (i.e. horizontal sharding) to be used
179 as persistence time.
180
181 This callable is not present on a plain :class:`.Session`, however
182 is established when using the horizontal sharding extension.
183
184 """
185
186 def __call__(
187 self,
188 mapper: Optional[Mapper[Any]] = None,
189 instance: Optional[object] = None,
190 **kw: Any,
191 ) -> Connection: ...
192
193
194def _state_session(state: InstanceState[Any]) -> Optional[Session]:
195 """Given an :class:`.InstanceState`, return the :class:`.Session`
196 associated, if any.
197 """
198 return state.session
199
200
201class _SessionClassMethods:
202 """Class-level methods for :class:`.Session`, :class:`.sessionmaker`."""
203
204 @classmethod
205 @util.deprecated(
206 "1.3",
207 "The :meth:`.Session.close_all` method is deprecated and will be "
208 "removed in a future release. Please refer to "
209 ":func:`.session.close_all_sessions`.",
210 )
211 def close_all(cls) -> None:
212 """Close *all* sessions in memory."""
213
214 close_all_sessions()
215
216 @classmethod
217 @util.preload_module("sqlalchemy.orm.util")
218 def identity_key(
219 cls,
220 class_: Optional[Type[Any]] = None,
221 ident: Union[Any, Tuple[Any, ...]] = None,
222 *,
223 instance: Optional[Any] = None,
224 row: Optional[Union[Row[Any], RowMapping]] = None,
225 identity_token: Optional[Any] = None,
226 ) -> _IdentityKeyType[Any]:
227 """Return an identity key.
228
229 This is an alias of :func:`.util.identity_key`.
230
231 """
232 return util.preloaded.orm_util.identity_key(
233 class_,
234 ident,
235 instance=instance,
236 row=row,
237 identity_token=identity_token,
238 )
239
240 @classmethod
241 def object_session(cls, instance: object) -> Optional[Session]:
242 """Return the :class:`.Session` to which an object belongs.
243
244 This is an alias of :func:`.object_session`.
245
246 """
247
248 return object_session(instance)
249
250
251class SessionTransactionState(_StateChangeState):
252 ACTIVE = 1
253 PREPARED = 2
254 COMMITTED = 3
255 DEACTIVE = 4
256 CLOSED = 5
257 PROVISIONING_CONNECTION = 6
258
259
260# backwards compatibility
261ACTIVE, PREPARED, COMMITTED, DEACTIVE, CLOSED, PROVISIONING_CONNECTION = tuple(
262 SessionTransactionState
263)
264
265
266class ORMExecuteState(util.MemoizedSlots):
267 """Represents a call to the :meth:`_orm.Session.execute` method, as passed
268 to the :meth:`.SessionEvents.do_orm_execute` event hook.
269
270 .. versionadded:: 1.4
271
272 .. seealso::
273
274 :ref:`session_execute_events` - top level documentation on how
275 to use :meth:`_orm.SessionEvents.do_orm_execute`
276
277 """
278
279 __slots__ = (
280 "session",
281 "statement",
282 "parameters",
283 "execution_options",
284 "local_execution_options",
285 "bind_arguments",
286 "identity_token",
287 "_compile_state_cls",
288 "_starting_event_idx",
289 "_events_todo",
290 "_update_execution_options",
291 )
292
293 session: Session
294 """The :class:`_orm.Session` in use."""
295
296 statement: Executable
297 """The SQL statement being invoked.
298
299 For an ORM selection as would
300 be retrieved from :class:`_orm.Query`, this is an instance of
301 :class:`_sql.select` that was generated from the ORM query.
302 """
303
304 parameters: Optional[_CoreAnyExecuteParams]
305 """Dictionary of parameters that was passed to
306 :meth:`_orm.Session.execute`."""
307
308 execution_options: _ExecuteOptions
309 """The complete dictionary of current execution options.
310
311 This is a merge of the statement level options with the
312 locally passed execution options.
313
314 .. seealso::
315
316 :attr:`_orm.ORMExecuteState.local_execution_options`
317
318 :meth:`_sql.Executable.execution_options`
319
320 :ref:`orm_queryguide_execution_options`
321
322 """
323
324 local_execution_options: _ExecuteOptions
325 """Dictionary view of the execution options passed to the
326 :meth:`.Session.execute` method.
327
328 This does not include options that may be associated with the statement
329 being invoked.
330
331 .. seealso::
332
333 :attr:`_orm.ORMExecuteState.execution_options`
334
335 """
336
337 bind_arguments: _BindArguments
338 """The dictionary passed as the
339 :paramref:`_orm.Session.execute.bind_arguments` dictionary.
340
341 This dictionary may be used by extensions to :class:`_orm.Session` to pass
342 arguments that will assist in determining amongst a set of database
343 connections which one should be used to invoke this statement.
344
345 """
346
347 _compile_state_cls: Optional[Type[ORMCompileState]]
348 _starting_event_idx: int
349 _events_todo: List[Any]
350 _update_execution_options: Optional[_ExecuteOptions]
351
352 def __init__(
353 self,
354 session: Session,
355 statement: Executable,
356 parameters: Optional[_CoreAnyExecuteParams],
357 execution_options: _ExecuteOptions,
358 bind_arguments: _BindArguments,
359 compile_state_cls: Optional[Type[ORMCompileState]],
360 events_todo: List[_InstanceLevelDispatch[Session]],
361 ):
362 """Construct a new :class:`_orm.ORMExecuteState`.
363
364 this object is constructed internally.
365
366 """
367 self.session = session
368 self.statement = statement
369 self.parameters = parameters
370 self.local_execution_options = execution_options
371 self.execution_options = statement._execution_options.union(
372 execution_options
373 )
374 self.bind_arguments = bind_arguments
375 self._compile_state_cls = compile_state_cls
376 self._events_todo = list(events_todo)
377
378 def _remaining_events(self) -> List[_InstanceLevelDispatch[Session]]:
379 return self._events_todo[self._starting_event_idx + 1 :]
380
381 def invoke_statement(
382 self,
383 statement: Optional[Executable] = None,
384 params: Optional[_CoreAnyExecuteParams] = None,
385 execution_options: Optional[OrmExecuteOptionsParameter] = None,
386 bind_arguments: Optional[_BindArguments] = None,
387 ) -> Result[Any]:
388 """Execute the statement represented by this
389 :class:`.ORMExecuteState`, without re-invoking events that have
390 already proceeded.
391
392 This method essentially performs a re-entrant execution of the current
393 statement for which the :meth:`.SessionEvents.do_orm_execute` event is
394 being currently invoked. The use case for this is for event handlers
395 that want to override how the ultimate
396 :class:`_engine.Result` object is returned, such as for schemes that
397 retrieve results from an offline cache or which concatenate results
398 from multiple executions.
399
400 When the :class:`_engine.Result` object is returned by the actual
401 handler function within :meth:`_orm.SessionEvents.do_orm_execute` and
402 is propagated to the calling
403 :meth:`_orm.Session.execute` method, the remainder of the
404 :meth:`_orm.Session.execute` method is preempted and the
405 :class:`_engine.Result` object is returned to the caller of
406 :meth:`_orm.Session.execute` immediately.
407
408 :param statement: optional statement to be invoked, in place of the
409 statement currently represented by :attr:`.ORMExecuteState.statement`.
410
411 :param params: optional dictionary of parameters or list of parameters
412 which will be merged into the existing
413 :attr:`.ORMExecuteState.parameters` of this :class:`.ORMExecuteState`.
414
415 .. versionchanged:: 2.0 a list of parameter dictionaries is accepted
416 for executemany executions.
417
418 :param execution_options: optional dictionary of execution options
419 will be merged into the existing
420 :attr:`.ORMExecuteState.execution_options` of this
421 :class:`.ORMExecuteState`.
422
423 :param bind_arguments: optional dictionary of bind_arguments
424 which will be merged amongst the current
425 :attr:`.ORMExecuteState.bind_arguments`
426 of this :class:`.ORMExecuteState`.
427
428 :return: a :class:`_engine.Result` object with ORM-level results.
429
430 .. seealso::
431
432 :ref:`do_orm_execute_re_executing` - background and examples on the
433 appropriate usage of :meth:`_orm.ORMExecuteState.invoke_statement`.
434
435
436 """
437
438 if statement is None:
439 statement = self.statement
440
441 _bind_arguments = dict(self.bind_arguments)
442 if bind_arguments:
443 _bind_arguments.update(bind_arguments)
444 _bind_arguments["_sa_skip_events"] = True
445
446 _params: Optional[_CoreAnyExecuteParams]
447 if params:
448 if self.is_executemany:
449 _params = []
450 exec_many_parameters = cast(
451 "List[Dict[str, Any]]", self.parameters
452 )
453 for _existing_params, _new_params in itertools.zip_longest(
454 exec_many_parameters,
455 cast("List[Dict[str, Any]]", params),
456 ):
457 if _existing_params is None or _new_params is None:
458 raise sa_exc.InvalidRequestError(
459 f"Can't apply executemany parameters to "
460 f"statement; number of parameter sets passed to "
461 f"Session.execute() ({len(exec_many_parameters)}) "
462 f"does not match number of parameter sets given "
463 f"to ORMExecuteState.invoke_statement() "
464 f"({len(params)})"
465 )
466 _existing_params = dict(_existing_params)
467 _existing_params.update(_new_params)
468 _params.append(_existing_params)
469 else:
470 _params = dict(cast("Dict[str, Any]", self.parameters))
471 _params.update(cast("Dict[str, Any]", params))
472 else:
473 _params = self.parameters
474
475 _execution_options = self.local_execution_options
476 if execution_options:
477 _execution_options = _execution_options.union(execution_options)
478
479 return self.session._execute_internal(
480 statement,
481 _params,
482 execution_options=_execution_options,
483 bind_arguments=_bind_arguments,
484 _parent_execute_state=self,
485 )
486
487 @property
488 def bind_mapper(self) -> Optional[Mapper[Any]]:
489 """Return the :class:`_orm.Mapper` that is the primary "bind" mapper.
490
491 For an :class:`_orm.ORMExecuteState` object invoking an ORM
492 statement, that is, the :attr:`_orm.ORMExecuteState.is_orm_statement`
493 attribute is ``True``, this attribute will return the
494 :class:`_orm.Mapper` that is considered to be the "primary" mapper
495 of the statement. The term "bind mapper" refers to the fact that
496 a :class:`_orm.Session` object may be "bound" to multiple
497 :class:`_engine.Engine` objects keyed to mapped classes, and the
498 "bind mapper" determines which of those :class:`_engine.Engine` objects
499 would be selected.
500
501 For a statement that is invoked against a single mapped class,
502 :attr:`_orm.ORMExecuteState.bind_mapper` is intended to be a reliable
503 way of getting this mapper.
504
505 .. versionadded:: 1.4.0b2
506
507 .. seealso::
508
509 :attr:`_orm.ORMExecuteState.all_mappers`
510
511
512 """
513 mp: Optional[Mapper[Any]] = self.bind_arguments.get("mapper", None)
514 return mp
515
516 @property
517 def all_mappers(self) -> Sequence[Mapper[Any]]:
518 """Return a sequence of all :class:`_orm.Mapper` objects that are
519 involved at the top level of this statement.
520
521 By "top level" we mean those :class:`_orm.Mapper` objects that would
522 be represented in the result set rows for a :func:`_sql.select`
523 query, or for a :func:`_dml.update` or :func:`_dml.delete` query,
524 the mapper that is the main subject of the UPDATE or DELETE.
525
526 .. versionadded:: 1.4.0b2
527
528 .. seealso::
529
530 :attr:`_orm.ORMExecuteState.bind_mapper`
531
532
533
534 """
535 if not self.is_orm_statement:
536 return []
537 elif isinstance(self.statement, (Select, FromStatement)):
538 result = []
539 seen = set()
540 for d in self.statement.column_descriptions:
541 ent = d["entity"]
542 if ent:
543 insp = inspect(ent, raiseerr=False)
544 if insp and insp.mapper and insp.mapper not in seen:
545 seen.add(insp.mapper)
546 result.append(insp.mapper)
547 return result
548 elif self.statement.is_dml and self.bind_mapper:
549 return [self.bind_mapper]
550 else:
551 return []
552
553 @property
554 def is_orm_statement(self) -> bool:
555 """return True if the operation is an ORM statement.
556
557 This indicates that the select(), insert(), update(), or delete()
558 being invoked contains ORM entities as subjects. For a statement
559 that does not have ORM entities and instead refers only to
560 :class:`.Table` metadata, it is invoked as a Core SQL statement
561 and no ORM-level automation takes place.
562
563 """
564 return self._compile_state_cls is not None
565
566 @property
567 def is_executemany(self) -> bool:
568 """return True if the parameters are a multi-element list of
569 dictionaries with more than one dictionary.
570
571 .. versionadded:: 2.0
572
573 """
574 return isinstance(self.parameters, list)
575
576 @property
577 def is_select(self) -> bool:
578 """return True if this is a SELECT operation.
579
580 .. versionchanged:: 2.0.30 - the attribute is also True for a
581 :meth:`_sql.Select.from_statement` construct that is itself against
582 a :class:`_sql.Select` construct, such as
583 ``select(Entity).from_statement(select(..))``
584
585 """
586 return self.statement.is_select
587
588 @property
589 def is_from_statement(self) -> bool:
590 """return True if this operation is a
591 :meth:`_sql.Select.from_statement` operation.
592
593 This is independent from :attr:`_orm.ORMExecuteState.is_select`, as a
594 ``select().from_statement()`` construct can be used with
595 INSERT/UPDATE/DELETE RETURNING types of statements as well.
596 :attr:`_orm.ORMExecuteState.is_select` will only be set if the
597 :meth:`_sql.Select.from_statement` is itself against a
598 :class:`_sql.Select` construct.
599
600 .. versionadded:: 2.0.30
601
602 """
603 return self.statement.is_from_statement
604
605 @property
606 def is_insert(self) -> bool:
607 """return True if this is an INSERT operation.
608
609 .. versionchanged:: 2.0.30 - the attribute is also True for a
610 :meth:`_sql.Select.from_statement` construct that is itself against
611 a :class:`_sql.Insert` construct, such as
612 ``select(Entity).from_statement(insert(..))``
613
614 """
615 return self.statement.is_dml and self.statement.is_insert
616
617 @property
618 def is_update(self) -> bool:
619 """return True if this is an UPDATE operation.
620
621 .. versionchanged:: 2.0.30 - the attribute is also True for a
622 :meth:`_sql.Select.from_statement` construct that is itself against
623 a :class:`_sql.Update` construct, such as
624 ``select(Entity).from_statement(update(..))``
625
626 """
627 return self.statement.is_dml and self.statement.is_update
628
629 @property
630 def is_delete(self) -> bool:
631 """return True if this is a DELETE operation.
632
633 .. versionchanged:: 2.0.30 - the attribute is also True for a
634 :meth:`_sql.Select.from_statement` construct that is itself against
635 a :class:`_sql.Delete` construct, such as
636 ``select(Entity).from_statement(delete(..))``
637
638 """
639 return self.statement.is_dml and self.statement.is_delete
640
641 @property
642 def _is_crud(self) -> bool:
643 return isinstance(self.statement, (dml.Update, dml.Delete))
644
645 def update_execution_options(self, **opts: Any) -> None:
646 """Update the local execution options with new values."""
647 self.local_execution_options = self.local_execution_options.union(opts)
648
649 def _orm_compile_options(
650 self,
651 ) -> Optional[
652 Union[
653 context.ORMCompileState.default_compile_options,
654 Type[context.ORMCompileState.default_compile_options],
655 ]
656 ]:
657 if not self.is_select:
658 return None
659 try:
660 opts = self.statement._compile_options
661 except AttributeError:
662 return None
663
664 if opts is not None and opts.isinstance(
665 context.ORMCompileState.default_compile_options
666 ):
667 return opts # type: ignore
668 else:
669 return None
670
671 @property
672 def lazy_loaded_from(self) -> Optional[InstanceState[Any]]:
673 """An :class:`.InstanceState` that is using this statement execution
674 for a lazy load operation.
675
676 The primary rationale for this attribute is to support the horizontal
677 sharding extension, where it is available within specific query
678 execution time hooks created by this extension. To that end, the
679 attribute is only intended to be meaningful at **query execution
680 time**, and importantly not any time prior to that, including query
681 compilation time.
682
683 """
684 return self.load_options._lazy_loaded_from
685
686 @property
687 def loader_strategy_path(self) -> Optional[PathRegistry]:
688 """Return the :class:`.PathRegistry` for the current load path.
689
690 This object represents the "path" in a query along relationships
691 when a particular object or collection is being loaded.
692
693 """
694 opts = self._orm_compile_options()
695 if opts is not None:
696 return opts._current_path
697 else:
698 return None
699
700 @property
701 def is_column_load(self) -> bool:
702 """Return True if the operation is refreshing column-oriented
703 attributes on an existing ORM object.
704
705 This occurs during operations such as :meth:`_orm.Session.refresh`,
706 as well as when an attribute deferred by :func:`_orm.defer` is
707 being loaded, or an attribute that was expired either directly
708 by :meth:`_orm.Session.expire` or via a commit operation is being
709 loaded.
710
711 Handlers will very likely not want to add any options to queries
712 when such an operation is occurring as the query should be a straight
713 primary key fetch which should not have any additional WHERE criteria,
714 and loader options travelling with the instance
715 will have already been added to the query.
716
717 .. versionadded:: 1.4.0b2
718
719 .. seealso::
720
721 :attr:`_orm.ORMExecuteState.is_relationship_load`
722
723 """
724 opts = self._orm_compile_options()
725 return opts is not None and opts._for_refresh_state
726
727 @property
728 def is_relationship_load(self) -> bool:
729 """Return True if this load is loading objects on behalf of a
730 relationship.
731
732 This means, the loader in effect is either a LazyLoader,
733 SelectInLoader, SubqueryLoader, or similar, and the entire
734 SELECT statement being emitted is on behalf of a relationship
735 load.
736
737 Handlers will very likely not want to add any options to queries
738 when such an operation is occurring, as loader options are already
739 capable of being propagated to relationship loaders and should
740 be already present.
741
742 .. seealso::
743
744 :attr:`_orm.ORMExecuteState.is_column_load`
745
746 """
747 opts = self._orm_compile_options()
748 if opts is None:
749 return False
750 path = self.loader_strategy_path
751 return path is not None and not path.is_root
752
753 @property
754 def load_options(
755 self,
756 ) -> Union[
757 context.QueryContext.default_load_options,
758 Type[context.QueryContext.default_load_options],
759 ]:
760 """Return the load_options that will be used for this execution."""
761
762 if not self.is_select:
763 raise sa_exc.InvalidRequestError(
764 "This ORM execution is not against a SELECT statement "
765 "so there are no load options."
766 )
767
768 lo: Union[
769 context.QueryContext.default_load_options,
770 Type[context.QueryContext.default_load_options],
771 ] = self.execution_options.get(
772 "_sa_orm_load_options", context.QueryContext.default_load_options
773 )
774 return lo
775
776 @property
777 def update_delete_options(
778 self,
779 ) -> Union[
780 bulk_persistence.BulkUDCompileState.default_update_options,
781 Type[bulk_persistence.BulkUDCompileState.default_update_options],
782 ]:
783 """Return the update_delete_options that will be used for this
784 execution."""
785
786 if not self._is_crud:
787 raise sa_exc.InvalidRequestError(
788 "This ORM execution is not against an UPDATE or DELETE "
789 "statement so there are no update options."
790 )
791 uo: Union[
792 bulk_persistence.BulkUDCompileState.default_update_options,
793 Type[bulk_persistence.BulkUDCompileState.default_update_options],
794 ] = self.execution_options.get(
795 "_sa_orm_update_options",
796 bulk_persistence.BulkUDCompileState.default_update_options,
797 )
798 return uo
799
800 @property
801 def _non_compile_orm_options(self) -> Sequence[ORMOption]:
802 return [
803 opt
804 for opt in self.statement._with_options
805 if is_orm_option(opt) and not opt._is_compile_state
806 ]
807
808 @property
809 def user_defined_options(self) -> Sequence[UserDefinedOption]:
810 """The sequence of :class:`.UserDefinedOptions` that have been
811 associated with the statement being invoked.
812
813 """
814 return [
815 opt
816 for opt in self.statement._with_options
817 if is_user_defined_option(opt)
818 ]
819
820
821class SessionTransactionOrigin(Enum):
822 """indicates the origin of a :class:`.SessionTransaction`.
823
824 This enumeration is present on the
825 :attr:`.SessionTransaction.origin` attribute of any
826 :class:`.SessionTransaction` object.
827
828 .. versionadded:: 2.0
829
830 """
831
832 AUTOBEGIN = 0
833 """transaction were started by autobegin"""
834
835 BEGIN = 1
836 """transaction were started by calling :meth:`_orm.Session.begin`"""
837
838 BEGIN_NESTED = 2
839 """tranaction were started by :meth:`_orm.Session.begin_nested`"""
840
841 SUBTRANSACTION = 3
842 """transaction is an internal "subtransaction" """
843
844
845class SessionTransaction(_StateChange, TransactionalContext):
846 """A :class:`.Session`-level transaction.
847
848 :class:`.SessionTransaction` is produced from the
849 :meth:`_orm.Session.begin`
850 and :meth:`_orm.Session.begin_nested` methods. It's largely an internal
851 object that in modern use provides a context manager for session
852 transactions.
853
854 Documentation on interacting with :class:`_orm.SessionTransaction` is
855 at: :ref:`unitofwork_transaction`.
856
857
858 .. versionchanged:: 1.4 The scoping and API methods to work with the
859 :class:`_orm.SessionTransaction` object directly have been simplified.
860
861 .. seealso::
862
863 :ref:`unitofwork_transaction`
864
865 :meth:`.Session.begin`
866
867 :meth:`.Session.begin_nested`
868
869 :meth:`.Session.rollback`
870
871 :meth:`.Session.commit`
872
873 :meth:`.Session.in_transaction`
874
875 :meth:`.Session.in_nested_transaction`
876
877 :meth:`.Session.get_transaction`
878
879 :meth:`.Session.get_nested_transaction`
880
881
882 """
883
884 _rollback_exception: Optional[BaseException] = None
885
886 _connections: Dict[
887 Union[Engine, Connection], Tuple[Connection, Transaction, bool, bool]
888 ]
889 session: Session
890 _parent: Optional[SessionTransaction]
891
892 _state: SessionTransactionState
893
894 _new: weakref.WeakKeyDictionary[InstanceState[Any], object]
895 _deleted: weakref.WeakKeyDictionary[InstanceState[Any], object]
896 _dirty: weakref.WeakKeyDictionary[InstanceState[Any], object]
897 _key_switches: weakref.WeakKeyDictionary[
898 InstanceState[Any], Tuple[Any, Any]
899 ]
900
901 origin: SessionTransactionOrigin
902 """Origin of this :class:`_orm.SessionTransaction`.
903
904 Refers to a :class:`.SessionTransactionOrigin` instance which is an
905 enumeration indicating the source event that led to constructing
906 this :class:`_orm.SessionTransaction`.
907
908 .. versionadded:: 2.0
909
910 """
911
912 nested: bool = False
913 """Indicates if this is a nested, or SAVEPOINT, transaction.
914
915 When :attr:`.SessionTransaction.nested` is True, it is expected
916 that :attr:`.SessionTransaction.parent` will be present as well,
917 linking to the enclosing :class:`.SessionTransaction`.
918
919 .. seealso::
920
921 :attr:`.SessionTransaction.origin`
922
923 """
924
925 def __init__(
926 self,
927 session: Session,
928 origin: SessionTransactionOrigin,
929 parent: Optional[SessionTransaction] = None,
930 ):
931 TransactionalContext._trans_ctx_check(session)
932
933 self.session = session
934 self._connections = {}
935 self._parent = parent
936 self.nested = nested = origin is SessionTransactionOrigin.BEGIN_NESTED
937 self.origin = origin
938
939 if session._close_state is _SessionCloseState.CLOSED:
940 raise sa_exc.InvalidRequestError(
941 "This Session has been permanently closed and is unable "
942 "to handle any more transaction requests."
943 )
944
945 if nested:
946 if not parent:
947 raise sa_exc.InvalidRequestError(
948 "Can't start a SAVEPOINT transaction when no existing "
949 "transaction is in progress"
950 )
951
952 self._previous_nested_transaction = session._nested_transaction
953 elif origin is SessionTransactionOrigin.SUBTRANSACTION:
954 assert parent is not None
955 else:
956 assert parent is None
957
958 self._state = SessionTransactionState.ACTIVE
959
960 self._take_snapshot()
961
962 # make sure transaction is assigned before we call the
963 # dispatch
964 self.session._transaction = self
965
966 self.session.dispatch.after_transaction_create(self.session, self)
967
968 def _raise_for_prerequisite_state(
969 self, operation_name: str, state: _StateChangeState
970 ) -> NoReturn:
971 if state is SessionTransactionState.DEACTIVE:
972 if self._rollback_exception:
973 raise sa_exc.PendingRollbackError(
974 "This Session's transaction has been rolled back "
975 "due to a previous exception during flush."
976 " To begin a new transaction with this Session, "
977 "first issue Session.rollback()."
978 f" Original exception was: {self._rollback_exception}",
979 code="7s2a",
980 )
981 else:
982 raise sa_exc.InvalidRequestError(
983 "This session is in 'inactive' state, due to the "
984 "SQL transaction being rolled back; no further SQL "
985 "can be emitted within this transaction."
986 )
987 elif state is SessionTransactionState.CLOSED:
988 raise sa_exc.ResourceClosedError("This transaction is closed")
989 elif state is SessionTransactionState.PROVISIONING_CONNECTION:
990 raise sa_exc.InvalidRequestError(
991 "This session is provisioning a new connection; concurrent "
992 "operations are not permitted",
993 code="isce",
994 )
995 else:
996 raise sa_exc.InvalidRequestError(
997 f"This session is in '{state.name.lower()}' state; no "
998 "further SQL can be emitted within this transaction."
999 )
1000
1001 @property
1002 def parent(self) -> Optional[SessionTransaction]:
1003 """The parent :class:`.SessionTransaction` of this
1004 :class:`.SessionTransaction`.
1005
1006 If this attribute is ``None``, indicates this
1007 :class:`.SessionTransaction` is at the top of the stack, and
1008 corresponds to a real "COMMIT"/"ROLLBACK"
1009 block. If non-``None``, then this is either a "subtransaction"
1010 (an internal marker object used by the flush process) or a
1011 "nested" / SAVEPOINT transaction. If the
1012 :attr:`.SessionTransaction.nested` attribute is ``True``, then
1013 this is a SAVEPOINT, and if ``False``, indicates this a subtransaction.
1014
1015 """
1016 return self._parent
1017
1018 @property
1019 def is_active(self) -> bool:
1020 return (
1021 self.session is not None
1022 and self._state is SessionTransactionState.ACTIVE
1023 )
1024
1025 @property
1026 def _is_transaction_boundary(self) -> bool:
1027 return self.nested or not self._parent
1028
1029 @_StateChange.declare_states(
1030 (SessionTransactionState.ACTIVE,), _StateChangeStates.NO_CHANGE
1031 )
1032 def connection(
1033 self,
1034 bindkey: Optional[Mapper[Any]],
1035 execution_options: Optional[_ExecuteOptions] = None,
1036 **kwargs: Any,
1037 ) -> Connection:
1038 bind = self.session.get_bind(bindkey, **kwargs)
1039 return self._connection_for_bind(bind, execution_options)
1040
1041 @_StateChange.declare_states(
1042 (SessionTransactionState.ACTIVE,), _StateChangeStates.NO_CHANGE
1043 )
1044 def _begin(self, nested: bool = False) -> SessionTransaction:
1045 return SessionTransaction(
1046 self.session,
1047 (
1048 SessionTransactionOrigin.BEGIN_NESTED
1049 if nested
1050 else SessionTransactionOrigin.SUBTRANSACTION
1051 ),
1052 self,
1053 )
1054
1055 def _iterate_self_and_parents(
1056 self, upto: Optional[SessionTransaction] = None
1057 ) -> Iterable[SessionTransaction]:
1058 current = self
1059 result: Tuple[SessionTransaction, ...] = ()
1060 while current:
1061 result += (current,)
1062 if current._parent is upto:
1063 break
1064 elif current._parent is None:
1065 raise sa_exc.InvalidRequestError(
1066 "Transaction %s is not on the active transaction list"
1067 % (upto)
1068 )
1069 else:
1070 current = current._parent
1071
1072 return result
1073
1074 def _take_snapshot(self) -> None:
1075 if not self._is_transaction_boundary:
1076 parent = self._parent
1077 assert parent is not None
1078 self._new = parent._new
1079 self._deleted = parent._deleted
1080 self._dirty = parent._dirty
1081 self._key_switches = parent._key_switches
1082 return
1083
1084 is_begin = self.origin in (
1085 SessionTransactionOrigin.BEGIN,
1086 SessionTransactionOrigin.AUTOBEGIN,
1087 )
1088 if not is_begin and not self.session._flushing:
1089 self.session.flush()
1090
1091 self._new = weakref.WeakKeyDictionary()
1092 self._deleted = weakref.WeakKeyDictionary()
1093 self._dirty = weakref.WeakKeyDictionary()
1094 self._key_switches = weakref.WeakKeyDictionary()
1095
1096 def _restore_snapshot(self, dirty_only: bool = False) -> None:
1097 """Restore the restoration state taken before a transaction began.
1098
1099 Corresponds to a rollback.
1100
1101 """
1102 assert self._is_transaction_boundary
1103
1104 to_expunge = set(self._new).union(self.session._new)
1105 self.session._expunge_states(to_expunge, to_transient=True)
1106
1107 for s, (oldkey, newkey) in self._key_switches.items():
1108 # we probably can do this conditionally based on
1109 # if we expunged or not, but safe_discard does that anyway
1110 self.session.identity_map.safe_discard(s)
1111
1112 # restore the old key
1113 s.key = oldkey
1114
1115 # now restore the object, but only if we didn't expunge
1116 if s not in to_expunge:
1117 self.session.identity_map.replace(s)
1118
1119 for s in set(self._deleted).union(self.session._deleted):
1120 self.session._update_impl(s, revert_deletion=True)
1121
1122 assert not self.session._deleted
1123
1124 for s in self.session.identity_map.all_states():
1125 if not dirty_only or s.modified or s in self._dirty:
1126 s._expire(s.dict, self.session.identity_map._modified)
1127
1128 def _remove_snapshot(self) -> None:
1129 """Remove the restoration state taken before a transaction began.
1130
1131 Corresponds to a commit.
1132
1133 """
1134 assert self._is_transaction_boundary
1135
1136 if not self.nested and self.session.expire_on_commit:
1137 for s in self.session.identity_map.all_states():
1138 s._expire(s.dict, self.session.identity_map._modified)
1139
1140 statelib.InstanceState._detach_states(
1141 list(self._deleted), self.session
1142 )
1143 self._deleted.clear()
1144 elif self.nested:
1145 parent = self._parent
1146 assert parent is not None
1147 parent._new.update(self._new)
1148 parent._dirty.update(self._dirty)
1149 parent._deleted.update(self._deleted)
1150 parent._key_switches.update(self._key_switches)
1151
1152 @_StateChange.declare_states(
1153 (SessionTransactionState.ACTIVE,), _StateChangeStates.NO_CHANGE
1154 )
1155 def _connection_for_bind(
1156 self,
1157 bind: _SessionBind,
1158 execution_options: Optional[CoreExecuteOptionsParameter],
1159 ) -> Connection:
1160 if bind in self._connections:
1161 if execution_options:
1162 util.warn(
1163 "Connection is already established for the "
1164 "given bind; execution_options ignored"
1165 )
1166 return self._connections[bind][0]
1167
1168 self._state = SessionTransactionState.PROVISIONING_CONNECTION
1169
1170 local_connect = False
1171 should_commit = True
1172
1173 try:
1174 if self._parent:
1175 conn = self._parent._connection_for_bind(
1176 bind, execution_options
1177 )
1178 if not self.nested:
1179 return conn
1180 else:
1181 if isinstance(bind, engine.Connection):
1182 conn = bind
1183 if conn.engine in self._connections:
1184 raise sa_exc.InvalidRequestError(
1185 "Session already has a Connection associated "
1186 "for the given Connection's Engine"
1187 )
1188 else:
1189 conn = bind.connect()
1190 local_connect = True
1191
1192 try:
1193 if execution_options:
1194 conn = conn.execution_options(**execution_options)
1195
1196 transaction: Transaction
1197 if self.session.twophase and self._parent is None:
1198 # TODO: shouldn't we only be here if not
1199 # conn.in_transaction() ?
1200 # if twophase is set and conn.in_transaction(), validate