Python 4542 - Improved sessions API

doc/changelog.rst

@@ -1,6 +1,15 @@
 Changelog
 =========
 
+Changes in Version 4.17.0 (2026/XX/XX)
+--------------------------------------
+
+PyMongo 4.17 brings a number of changes including:
+
+- Added the :meth:`~pymongo.asynchronous.client_session.AsyncClientSession.bind` and :meth:`~pymongo.client_session.ClientSession.bind` methods
+  that allow users to bind a session to all database operations within the scope of a context manager instead of having to explicitly pass the session to each individual operation.
+  See <PLACEHOLDER> for examples and more information.
+
 Changes in Version 4.16.0 (2026/01/07)
 --------------------------------------
 

pymongo/asynchronous/client_session.py

@@ -139,6 +139,7 @@
 import time
 import uuid
 from collections.abc import Mapping as _Mapping
+from contextvars import ContextVar
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -153,6 +154,8 @@
     TypeVar,
 )
 
+from _contextvars import Token
+
 from bson.binary import Binary
 from bson.int64 import Int64
 from bson.timestamp import Timestamp
@@ -181,6 +184,14 @@
 
 _IS_SYNC = False
 
+_SESSION: ContextVar[Optional[_AsyncBoundClientSession]] = ContextVar("SESSION", default=None)
+
+
+class _AsyncBoundClientSession:
+    def __init__(self, session: AsyncClientSession, client_id: int):
+        self.session = session
+        self.client_id = client_id
+
 
 class SessionOptions:
     """Options for a new :class:`AsyncClientSession`.
@@ -517,6 +528,9 @@ def __init__(
         self._attached_to_cursor = False
         # Should we leave the session alive when the cursor is closed?
         self._leave_alive = False
+        # Is this session bound to a context manager scope?
+        self._bound = False
+        self._session_token: Optional[Token[_AsyncBoundClientSession]] = None
 
     async def end_session(self) -> None:
         """Finish this session. If a transaction has started, abort it.
@@ -547,11 +561,23 @@ def _check_ended(self) -> None:
         if self._server_session is None:
             raise InvalidOperation("Cannot use ended session")
 
+    def bind(self) -> AsyncClientSession:
+        self._bound = True
+        return self
+
     async def __aenter__(self) -> AsyncClientSession:
+        if self._bound:
+            bound_session = _AsyncBoundClientSession(self, id(self._client))
+            self._session_token = _SESSION.set(bound_session)  # type: ignore[assignment]
         return self
 
     async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
-        await self._end_session(lock=True)
+        if self._session_token:
+            _SESSION.reset(self._session_token)  # type: ignore[arg-type]
+            self._session_token = None
+            self._bound = False
+        else:
+            await self._end_session(lock=True)
 
     @property
     def client(self) -> AsyncMongoClient[Any]:

pymongo/asynchronous/mongo_client.py

@@ -65,7 +65,7 @@
 from pymongo.asynchronous import client_session, database, uri_parser
 from pymongo.asynchronous.change_stream import AsyncChangeStream, AsyncClusterChangeStream
 from pymongo.asynchronous.client_bulk import _AsyncClientBulk
-from pymongo.asynchronous.client_session import _EmptyServerSession
+from pymongo.asynchronous.client_session import _SESSION, _EmptyServerSession
 from pymongo.asynchronous.command_cursor import AsyncCommandCursor
 from pymongo.asynchronous.settings import TopologySettings
 from pymongo.asynchronous.topology import Topology, _ErrorContext
@@ -1408,7 +1408,8 @@ def start_session(
     def _ensure_session(
         self, session: Optional[AsyncClientSession] = None
     ) -> Optional[AsyncClientSession]:
-        """If provided session is None, lend a temporary session."""
+        """If provided session and bound session are None, lend a temporary session."""
+        session = session or self._get_bound_session()
         if session:
             return session
 
@@ -2267,11 +2268,14 @@ async def _tmp_session(
         self, session: Optional[client_session.AsyncClientSession]
     ) -> AsyncGenerator[Optional[client_session.AsyncClientSession], None]:
         """If provided session is None, lend a temporary session."""
-        if session is not None:
-            if not isinstance(session, client_session.AsyncClientSession):
-                raise ValueError(
-                    f"'session' argument must be an AsyncClientSession or None, not {type(session)}"
-                )
+        if session is not None and not isinstance(session, client_session.AsyncClientSession):
+            raise ValueError(
+                f"'session' argument must be an AsyncClientSession or None, not {type(session)}"
+            )
+
+        # Check for a bound session. If one exists, treat it as an explicitly passed session.
+        session = session or self._get_bound_session()
+        if session:
             # Don't call end_session.
             yield session
             return
@@ -2301,6 +2305,18 @@ async def _process_response(
         if session is not None:
             session._process_response(reply)
 
+    def _get_bound_session(self) -> Optional[AsyncClientSession]:
+        bound_session = _SESSION.get()
+        if bound_session:
+            if bound_session.client_id == id(self):
+                return bound_session.session
+            else:
+                raise InvalidOperation(
+                    "Only the client that created the bound session can perform operations within its context block. See <PLACEHOLDER> for more information."
+                )
+        else:
+            return None
+
     async def server_info(
         self, session: Optional[client_session.AsyncClientSession] = None
     ) -> dict[str, Any]:

pymongo/synchronous/client_session.py

@@ -139,6 +139,7 @@
 import time
 import uuid
 from collections.abc import Mapping as _Mapping
+from contextvars import ContextVar
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -152,6 +153,8 @@
     TypeVar,
 )
 
+from _contextvars import Token
+
 from bson.binary import Binary
 from bson.int64 import Int64
 from bson.timestamp import Timestamp
@@ -180,6 +183,14 @@
 
 _IS_SYNC = True
 
+_SESSION: ContextVar[Optional[_BoundClientSession]] = ContextVar("SESSION", default=None)
+
+
+class _BoundClientSession:
+    def __init__(self, session: ClientSession, client_id: int):
+        self.session = session
+        self.client_id = client_id
+
 
 class SessionOptions:
     """Options for a new :class:`ClientSession`.
@@ -516,6 +527,9 @@ def __init__(
         self._attached_to_cursor = False
         # Should we leave the session alive when the cursor is closed?
         self._leave_alive = False
+        # Is this session bound to a context manager scope?
+        self._bound = False
+        self._session_token: Optional[Token[_BoundClientSession]] = None
 
     def end_session(self) -> None:
         """Finish this session. If a transaction has started, abort it.
@@ -546,11 +560,23 @@ def _check_ended(self) -> None:
         if self._server_session is None:
             raise InvalidOperation("Cannot use ended session")
 
+    def bind(self) -> ClientSession:
+        self._bound = True
+        return self
+
     def __enter__(self) -> ClientSession:
+        if self._bound:
+            bound_session = _BoundClientSession(self, id(self._client))
+            self._session_token = _SESSION.set(bound_session)  # type: ignore[assignment]
         return self
 
     def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
-        self._end_session(lock=True)
+        if self._session_token:
+            _SESSION.reset(self._session_token)  # type: ignore[arg-type]
+            self._session_token = None
+            self._bound = False
+        else:
+            self._end_session(lock=True)
 
     @property
     def client(self) -> MongoClient[Any]:

pymongo/synchronous/mongo_client.py

@@ -108,7 +108,7 @@
 from pymongo.synchronous import client_session, database, uri_parser
 from pymongo.synchronous.change_stream import ChangeStream, ClusterChangeStream
 from pymongo.synchronous.client_bulk import _ClientBulk
-from pymongo.synchronous.client_session import _EmptyServerSession
+from pymongo.synchronous.client_session import _SESSION, _EmptyServerSession
 from pymongo.synchronous.command_cursor import CommandCursor
 from pymongo.synchronous.settings import TopologySettings
 from pymongo.synchronous.topology import Topology, _ErrorContext
@@ -1406,7 +1406,8 @@ def start_session(
         )
 
     def _ensure_session(self, session: Optional[ClientSession] = None) -> Optional[ClientSession]:
-        """If provided session is None, lend a temporary session."""
+        """If provided session and bound session are None, lend a temporary session."""
+        session = session or self._get_bound_session()
         if session:
             return session
 
@@ -2263,11 +2264,14 @@ def _tmp_session(
         self, session: Optional[client_session.ClientSession]
     ) -> Generator[Optional[client_session.ClientSession], None]:
         """If provided session is None, lend a temporary session."""
-        if session is not None:
-            if not isinstance(session, client_session.ClientSession):
-                raise ValueError(
-                    f"'session' argument must be a ClientSession or None, not {type(session)}"
-                )
+        if session is not None and not isinstance(session, client_session.ClientSession):
+            raise ValueError(
+                f"'session' argument must be a ClientSession or None, not {type(session)}"
+            )
+
+        # Check for a bound session. If one exists, treat it as an explicitly passed session.
+        session = session or self._get_bound_session()
+        if session:
             # Don't call end_session.
             yield session
             return
@@ -2295,6 +2299,18 @@ def _process_response(self, reply: Mapping[str, Any], session: Optional[ClientSe
         if session is not None:
             session._process_response(reply)
 
+    def _get_bound_session(self) -> Optional[ClientSession]:
+        bound_session = _SESSION.get()
+        if bound_session:
+            if bound_session.client_id == id(self):
+                return bound_session.session
+            else:
+                raise InvalidOperation(
+                    "Only the client that created the bound session can perform operations within its context block. See <PLACEHOLDER> for more information."
+                )
+        else:
+            return None
+
     def server_info(self, session: Optional[client_session.ClientSession] = None) -> dict[str, Any]:
         """Get information about the MongoDB server we're connected to.
 

test/asynchronous/test_session.py

@@ -189,6 +189,52 @@ async def _test_ops(self, client, *ops):
                         f"{f.__name__} did not return implicit session to pool",
                     )
 
+        # Explicit bound session
+        for f, args, kw in ops:
+            async with client.start_session() as s:
+                async with s.bind():
+                    listener.reset()
+                    s._materialize()
+                    last_use = s._server_session.last_use
+                    start = time.monotonic()
+                    self.assertLessEqual(last_use, start)
+                    # In case "f" modifies its inputs.
+                    args = copy.copy(args)
+                    kw = copy.copy(kw)
+                    await f(*args, **kw)
+                    self.assertGreaterEqual(len(listener.started_events), 1)
+                    for event in listener.started_events:
+                        self.assertIn(
+                            "lsid",
+                            event.command,
+                            f"{f.__name__} sent no lsid with {event.command_name}",
+                        )
+
+                        self.assertEqual(
+                            s.session_id,
+                            event.command["lsid"],
+                            f"{f.__name__} sent wrong lsid with {event.command_name}",
+                        )
+
+                    self.assertFalse(s.has_ended)
+
+            self.assertTrue(s.has_ended)
+            with self.assertRaisesRegex(InvalidOperation, "ended session"):
+                async with s.bind():
+                    await f(*args, **kw)
+
+            # Test a session cannot be used on another client.
+            async with self.client2.start_session() as s:
+                async with s.bind():
+                    # In case "f" modifies its inputs.
+                    args = copy.copy(args)
+                    kw = copy.copy(kw)
+                    with self.assertRaisesRegex(
+                        InvalidOperation,
+                        "Only the client that created the bound session can perform operations within its context block",
+                    ):
+                        await f(*args, **kw)
+
     async def test_implicit_sessions_checkout(self):
         # "To confirm that implicit sessions only allocate their server session after a
         # successful connection checkout" test from Driver Sessions Spec.
@@ -825,6 +871,57 @@ async def test_session_not_copyable(self):
         async with client.start_session() as s:
             self.assertRaises(TypeError, lambda: copy.copy(s))
 
+    async def test_nested_session_binding(self):
+        coll = self.client.pymongo_test.test
+        await coll.insert_one({"x": 1})
+
+        session1 = self.client.start_session()
+        session2 = self.client.start_session()
+        session1._materialize()
+        session2._materialize()
+        try:
+            self.listener.reset()
+            # Uses implicit session
+            await coll.find_one()
+            implicit_lsid = self.listener.started_events[0].command.get("lsid")
+            self.assertIsNotNone(implicit_lsid)
+            self.assertNotEqual(implicit_lsid, session1.session_id)
+            self.assertNotEqual(implicit_lsid, session2.session_id)
+
+            async with session1.bind():
+                self.listener.reset()
+                # Uses bound session1
+                await coll.find_one()
+                session1_lsid = self.listener.started_events[0].command.get("lsid")
+                self.assertEqual(session1_lsid, session1.session_id)
+
+                async with session2.bind():
+                    self.listener.reset()
+                    # Uses bound session2
+                    await coll.find_one()
+                    session2_lsid = self.listener.started_events[0].command.get("lsid")
+                    self.assertEqual(session2_lsid, session2.session_id)
+                    self.assertNotEqual(session2_lsid, session1.session_id)
+
+                self.listener.reset()
+                # Use bound session1 again
+                await coll.find_one()
+                session1_lsid = self.listener.started_events[0].command.get("lsid")
+                self.assertEqual(session1_lsid, session1.session_id)
+                self.assertNotEqual(session1_lsid, session2.session_id)
+
+            self.listener.reset()
+            # Uses implicit session
+            await coll.find_one()
+            implicit_lsid = self.listener.started_events[0].command.get("lsid")
+            self.assertIsNotNone(implicit_lsid)
+            self.assertNotEqual(implicit_lsid, session1.session_id)
+            self.assertNotEqual(implicit_lsid, session2.session_id)
+
+        finally:
+            await session1.end_session()
+            await session2.end_session()
+
 
 class TestCausalConsistency(AsyncUnitTest):
     listener: SessionTestListener