PYTHON-4747 Sync client_session.py to master

Author: blink1073

pymongo/client_session.py

@@ -0,0 +1,22 @@
+# Copyright 2024-present MongoDB, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Re-import of synchronous ClientSession API for compatibility."""
+from __future__ import annotations
+
+from pymongo.synchronous.client_session import *  # noqa: F403
+from pymongo.synchronous.client_session import __doc__ as original_doc
+
+__doc__ = original_doc
+__all__ = ["ClientSession", "SessionOptions", "TransactionOptions"]  # noqa: F405

pymongo/synchronous/client_session.py

@@ -156,7 +156,6 @@
 from bson.int64 import Int64
 from bson.timestamp import Timestamp
 from pymongo import _csot
-from pymongo.cursor import _ConnectionManager
 from pymongo.errors import (
     ConfigurationError,
     ConnectionFailure,
@@ -165,20 +164,23 @@
     PyMongoError,
     WTimeoutError,
 )
-from pymongo.helpers import _RETRYABLE_ERROR_CODES
+from pymongo.helpers_shared import _RETRYABLE_ERROR_CODES
 from pymongo.operations import _Op
 from pymongo.read_concern import ReadConcern
 from pymongo.read_preferences import ReadPreference, _ServerMode
 from pymongo.server_type import SERVER_TYPE
+from pymongo.synchronous.cursor import _ConnectionManager
 from pymongo.write_concern import WriteConcern
 
 if TYPE_CHECKING:
     from types import TracebackType
 
-    from pymongo.pool import Connection
-    from pymongo.server import Server
+    from pymongo.synchronous.pool import Connection
+    from pymongo.synchronous.server import Server
     from pymongo.typings import ClusterTime, _Address
 
+_IS_SYNC = True
+
 
 class SessionOptions:
     """Options for a new :class:`ClientSession`.
@@ -476,7 +478,7 @@ def _within_time_limit(start_time: float) -> bool:
 _T = TypeVar("_T")
 
 if TYPE_CHECKING:
-    from pymongo.mongo_client import MongoClient
+    from pymongo.synchronous.mongo_client import MongoClient
 
 
 class ClientSession:
@@ -526,9 +528,15 @@ def _end_session(self, lock: bool) -> None:
                 # is in the committed state when the session is discarded.
                 self._unpin()
             finally:
-                self._client._return_server_session(self._server_session, lock)
+                self._client._return_server_session(self._server_session)
                 self._server_session = None
 
+    def _end_implicit_session(self) -> None:
+        # Implicit sessions can't be part of transactions or pinned connections
+        if self._server_session is not None:
+            self._client._return_server_session(self._server_session)
+            self._server_session = None
+
     def _check_ended(self) -> None:
         if self._server_session is None:
             raise InvalidOperation("Cannot use ended session")
@@ -1097,7 +1105,7 @@ def inc_transaction_id(self) -> None:
 class _ServerSessionPool(collections.deque):
     """Pool of _ServerSession objects.
 
-    This class is not thread-safe, access it while holding the Topology lock.
+    This class is thread-safe.
     """
 
     def __init__(self, *args: Any, **kwargs: Any):
@@ -1110,8 +1118,11 @@ def reset(self) -> None:
 
     def pop_all(self) -> list[_ServerSession]:
         ids = []
-        while self:
-            ids.append(self.pop().session_id)
+        while True:
+            try:
+                ids.append(self.pop().session_id)
+            except IndexError:
+                break
         return ids
 
     def get_server_session(self, session_timeout_minutes: Optional[int]) -> _ServerSession:
@@ -1123,33 +1134,30 @@ def get_server_session(self, session_timeout_minutes: Optional[int]) -> _ServerS
         self._clear_stale(session_timeout_minutes)
 
         # The most recently used sessions are on the left.
-        while self:
-            s = self.popleft()
+        while True:
+            try:
+                s = self.popleft()
+            except IndexError:
+                break
             if not s.timed_out(session_timeout_minutes):
                 return s
 
         return _ServerSession(self.generation)
 
-    def return_server_session(
-        self, server_session: _ServerSession, session_timeout_minutes: Optional[int]
-    ) -> None:
-        if session_timeout_minutes is not None:
-            self._clear_stale(session_timeout_minutes)
-            if server_session.timed_out(session_timeout_minutes):
-                return
-        self.return_server_session_no_lock(server_session)
-
-    def return_server_session_no_lock(self, server_session: _ServerSession) -> None:
+    def return_server_session(self, server_session: _ServerSession) -> None:
         # Discard sessions from an old pool to avoid duplicate sessions in the
         # child process after a fork.
         if server_session.generation == self.generation and not server_session.dirty:
             self.appendleft(server_session)
 
     def _clear_stale(self, session_timeout_minutes: Optional[int]) -> None:
         # Clear stale sessions. The least recently used are on the right.
-        while self:
-            if self[-1].timed_out(session_timeout_minutes):
-                self.pop()
-            else:
+        while True:
+            try:
+                s = self.pop()
+            except IndexError:
+                break
+            if not s.timed_out(session_timeout_minutes):
+                self.append(s)
                 # The remaining sessions also haven't timed out.
                 break