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