implement autobegin=False option

Added new parameter :paramref:`_orm.Session.autobegin`, which when set to
``False`` will prevent the :class:`_orm.Session` from beginning a
transaction implicitly. The :meth:`_orm.Session.begin` method must be
called explicitly first in order to proceed with operations, otherwise an
error is raised whenever any operation would otherwise have begun
automatically. This option can be used to create a "safe"
:class:`_orm.Session` that won't implicitly start new transactions.

As part of this change, also added a new status variable
:class:`_orm.SessionTransaction.origin` which may be useful for event
handling code to be aware of the origin of a particular
:class:`_orm.SessionTransaction`.

Fixes: #6928
Change-Id: I246f895c4a475bff352216e5bc74b6a25e6a4ae7

3 files changed, 127 insertions, 51 deletions

diff --git a/lib/sqlalchemy/orm/__init__.py b/lib/sqlalchemy/orm/__init__.py
index c6b61f3b4..5e2161515 100644
--- a/lib/sqlalchemy/orm/__init__.py
+++ b/lib/sqlalchemy/orm/__init__.py
@@ -127,6 +127,7 @@ from .session import ORMExecuteState as ORMExecuteState
 from .session import Session as Session
 from .session import sessionmaker as sessionmaker
 from .session import SessionTransaction as SessionTransaction
+from .session import SessionTransactionOrigin as SessionTransactionOrigin
 from .state import AttributeState as AttributeState
 from .state import InstanceState as InstanceState
 from .strategy_options import contains_eager as contains_eager
diff --git a/lib/sqlalchemy/orm/scoping.py b/lib/sqlalchemy/orm/scoping.py
index e3ba86c5a..1971caedc 100644
--- a/lib/sqlalchemy/orm/scoping.py
+++ b/lib/sqlalchemy/orm/scoping.py
@@ -382,9 +382,7 @@ class scoped_session(Generic[_S]):
 
         return self._proxied.add_all(instances)
 
-    def begin(
-        self, nested: bool = False, _subtrans: bool = False
-    ) -> SessionTransaction:
+    def begin(self, nested: bool = False) -> SessionTransaction:
         r"""Begin a transaction, or nested transaction,
         on this :class:`.Session`, if one is not already begun.
 
@@ -425,7 +423,7 @@ class scoped_session(Generic[_S]):
 
         """  # noqa: E501
 
-        return self._proxied.begin(nested=nested, _subtrans=_subtrans)
+        return self._proxied.begin(nested=nested)
 
     def begin_nested(self) -> SessionTransaction:
         r"""Begin a "nested" transaction on this Session, e.g. SAVEPOINT.
@@ -1772,6 +1770,11 @@ class scoped_session(Generic[_S]):
 
         .. versionadded:: 1.4.24
 
+        .. seealso::
+
+            :ref:`orm_queryguide_select_orm_entities` - contrasts the behavior
+            of :meth:`_orm.Session.execute` to :meth:`_orm.Session.scalars`
+
 
         """  # noqa: E501
 
diff --git a/lib/sqlalchemy/orm/session.py b/lib/sqlalchemy/orm/session.py
index c759f6541..22e47585d 100644
--- a/lib/sqlalchemy/orm/session.py
+++ b/lib/sqlalchemy/orm/session.py
@@ -10,6 +10,7 @@
 from __future__ import annotations
 
 import contextlib
+from enum import Enum
 import itertools
 import sys
 import typing
@@ -734,6 +735,30 @@ class ORMExecuteState(util.MemoizedSlots):
         ]
 
 
+class SessionTransactionOrigin(Enum):
+    """indicates the origin of a :class:`.SessionTransaction`.
+
+    This enumeration is present on the
+    :attr:`.SessionTransaction.origin` attribute of any
+    :class:`.SessionTransaction` object.
+
+    .. versionadded:: 2.0
+
+    """
+
+    AUTOBEGIN = 0
+    """transaction were started by autobegin"""
+
+    BEGIN = 1
+    """transaction were started by calling :meth:`_orm.Session.begin`"""
+
+    BEGIN_NESTED = 2
+    """tranaction were started by :meth:`_orm.Session.begin_nested`"""
+
+    SUBTRANSACTION = 3
+    """transaction is an internal "subtransaction" """
+
+
 class SessionTransaction(_StateChange, TransactionalContext):
     """A :class:`.Session`-level transaction.
 
@@ -790,29 +815,60 @@ class SessionTransaction(_StateChange, TransactionalContext):
         InstanceState[Any], Tuple[Any, Any]
     ]
 
+    origin: SessionTransactionOrigin
+    """Origin of this :class:`_orm.SessionTransaction`.
+
+    Refers to a :class:`.SessionTransactionOrigin` instance which is an
+    enumeration indicating the source event that led to constructing
+    this :class:`_orm.SessionTransaction`.
+
+    .. versionadded:: 2.0
+
+    """
+
+    nested: bool = False
+    """Indicates if this is a nested, or SAVEPOINT, transaction.
+
+    When :attr:`.SessionTransaction.nested` is True, it is expected
+    that :attr:`.SessionTransaction.parent` will be present as well,
+    linking to the enclosing :class:`.SessionTransaction`.
+
+    .. seealso::
+
+        :attr:`.SessionTransaction.origin`
+
+    """
+
     def __init__(
         self,
         session: Session,
+        origin: SessionTransactionOrigin,
         parent: Optional[SessionTransaction] = None,
-        nested: bool = False,
-        autobegin: bool = False,
     ):
         TransactionalContext._trans_ctx_check(session)
 
         self.session = session
         self._connections = {}
         self._parent = parent
-        self.nested = nested
+        self.nested = nested = origin is SessionTransactionOrigin.BEGIN_NESTED
+        self.origin = origin
+
         if nested:
+            if not parent:
+                raise sa_exc.InvalidRequestError(
+                    "Can't start a SAVEPOINT transaction when no existing "
+                    "transaction is in progress"
+                )
+
             self._previous_nested_transaction = session._nested_transaction
+        elif origin is SessionTransactionOrigin.SUBTRANSACTION:
+            assert parent is not None
+        else:
+            assert parent is None
+
         self._state = SessionTransactionState.ACTIVE
-        if not parent and nested:
-            raise sa_exc.InvalidRequestError(
-                "Can't start a SAVEPOINT transaction when no existing "
-                "transaction is in progress"
-            )
 
-        self._take_snapshot(autobegin=autobegin)
+        self._take_snapshot()
 
         # make sure transaction is assigned before we call the
         # dispatch
@@ -866,14 +922,6 @@ class SessionTransaction(_StateChange, TransactionalContext):
         """
         return self._parent
 
-    nested: bool = False
-    """Indicates if this is a nested, or SAVEPOINT, transaction.
-
-    When :attr:`.SessionTransaction.nested` is True, it is expected
-    that :attr:`.SessionTransaction.parent` will be True as well.
-
-    """
-
     @property
     def is_active(self) -> bool:
         return (
@@ -901,7 +949,13 @@ class SessionTransaction(_StateChange, TransactionalContext):
         (SessionTransactionState.ACTIVE,), _StateChangeStates.NO_CHANGE
     )
     def _begin(self, nested: bool = False) -> SessionTransaction:
-        return SessionTransaction(self.session, self, nested=nested)
+        return SessionTransaction(
+            self.session,
+            SessionTransactionOrigin.BEGIN_NESTED
+            if nested
+            else SessionTransactionOrigin.SUBTRANSACTION,
+            self,
+        )
 
     def _iterate_self_and_parents(
         self, upto: Optional[SessionTransaction] = None
@@ -923,7 +977,7 @@ class SessionTransaction(_StateChange, TransactionalContext):
 
         return result
 
-    def _take_snapshot(self, autobegin: bool = False) -> None:
+    def _take_snapshot(self) -> None:
         if not self._is_transaction_boundary:
             parent = self._parent
             assert parent is not None
@@ -933,7 +987,11 @@ class SessionTransaction(_StateChange, TransactionalContext):
             self._key_switches = parent._key_switches
             return
 
-        if not autobegin and not self.session._flushing:
+        is_begin = self.origin in (
+            SessionTransactionOrigin.BEGIN,
+            SessionTransactionOrigin.AUTOBEGIN,
+        )
+        if not is_begin and not self.session._flushing:
             self.session.flush()
 
         self._new = weakref.WeakKeyDictionary()
@@ -1307,6 +1365,7 @@ class Session(_SessionClassMethods, EventTarget):
         autoflush: bool = True,
         future: Literal[True] = True,
         expire_on_commit: bool = True,
+        autobegin: bool = True,
         twophase: bool = False,
         binds: Optional[Dict[_SessionBindKey, _SessionBind]] = None,
         enable_baked_queries: bool = True,
@@ -1330,6 +1389,20 @@ class Session(_SessionClassMethods, EventTarget):
 
                :ref:`session_flushing` - additional background on autoflush
 
+        :param autobegin: Automatically start transactions (i.e. equivalent to
+           invoking :meth:`_orm.Session.begin`) when database access is
+           requested by an operation.   Defaults to ``True``.    Set to
+           ``False`` to prevent a :class:`_orm.Session` from implicitly
+           beginning transactions after construction, as well as after any of
+           the :meth:`_orm.Session.rollback`, :meth:`_orm.Session.commit`,
+           or :meth:`_orm.Session.close` methods are called.
+
+           .. versionadded:: 2.0
+
+           .. seealso::
+
+                :ref:`session_autobegin_disable`
+
         :param bind: An optional :class:`_engine.Engine` or
            :class:`_engine.Connection` to
            which this ``Session`` should be bound. When specified, all SQL
@@ -1455,6 +1528,7 @@ class Session(_SessionClassMethods, EventTarget):
         self._transaction = None
         self._nested_transaction = None
         self.hash_key = _new_sessionid()
+        self.autobegin = autobegin
         self.autoflush = autoflush
         self.expire_on_commit = expire_on_commit
         self.enable_baked_queries = enable_baked_queries
@@ -1542,18 +1616,26 @@ class Session(_SessionClassMethods, EventTarget):
         """
         return {}
 
-    def _autobegin_t(self) -> SessionTransaction:
+    def _autobegin_t(self, begin: bool = False) -> SessionTransaction:
         if self._transaction is None:
 
-            trans = SessionTransaction(self, autobegin=True)
+            if not begin and not self.autobegin:
+                raise sa_exc.InvalidRequestError(
+                    "Autobegin is disabled on this Session; please call "
+                    "session.begin() to start a new transaction"
+                )
+            trans = SessionTransaction(
+                self,
+                SessionTransactionOrigin.BEGIN
+                if begin
+                else SessionTransactionOrigin.AUTOBEGIN,
+            )
             assert self._transaction is trans
             return trans
 
         return self._transaction
 
-    def begin(
-        self, nested: bool = False, _subtrans: bool = False
-    ) -> SessionTransaction:
+    def begin(self, nested: bool = False) -> SessionTransaction:
         """Begin a transaction, or nested transaction,
         on this :class:`.Session`, if one is not already begun.
 
@@ -1590,31 +1672,21 @@ class Session(_SessionClassMethods, EventTarget):
 
         trans = self._transaction
         if trans is None:
-            trans = self._autobegin_t()
+            trans = self._autobegin_t(begin=True)
 
-            if not nested and not _subtrans:
+            if not nested:
                 return trans
 
-        if trans is not None:
-            if _subtrans or nested:
-                trans = trans._begin(nested=nested)
-                assert self._transaction is trans
-                if nested:
-                    self._nested_transaction = trans
-            else:
-                raise sa_exc.InvalidRequestError(
-                    "A transaction is already begun on this Session."
-                )
-        else:
-            # outermost transaction.  must be a not nested and not
-            # a subtransaction
+        assert trans is not None
 
-            assert not nested and not _subtrans
-            trans = SessionTransaction(self)
+        if nested:
+            trans = trans._begin(nested=nested)
             assert self._transaction is trans
-
-            if TYPE_CHECKING:
-                assert self._transaction is not None
+            self._nested_transaction = trans
+        else:
+            raise sa_exc.InvalidRequestError(
+                "A transaction is already begun on this Session."
+            )
 
         return trans  # needed for __enter__/__exit__ hook
 
@@ -3957,7 +4029,7 @@ class Session(_SessionClassMethods, EventTarget):
         if not flush_context.has_work:
             return
 
-        flush_context.transaction = transaction = self.begin(_subtrans=True)
+        flush_context.transaction = transaction = self._autobegin_t()._begin()
         try:
             self._warn_on_events = True
             try:
@@ -4246,7 +4318,7 @@ class Session(_SessionClassMethods, EventTarget):
         mapper = _class_to_mapper(mapper)
         self._flushing = True
 
-        transaction = self.begin(_subtrans=True)
+        transaction = self._autobegin_t()._begin()
         try:
             if isupdate:
                 bulk_persistence._bulk_update(