diff --git a/pymongo/asynchronous/bulk.py b/pymongo/asynchronous/bulk.py index 622b3cf2b7..86b9528b6e 100644 --- a/pymongo/asynchronous/bulk.py +++ b/pymongo/asynchronous/bulk.py @@ -33,7 +33,7 @@ from bson.raw_bson import RawBSONDocument from pymongo import _csot, common from pymongo._telemetry import _generate_op_id_or_none -from pymongo.asynchronous.client_session import AsyncClientSession, _validate_session_write_concern +from pymongo.asynchronous.client_session import AsyncClientSession from pymongo.asynchronous.command_runner import ( run_bulk_write_command, ) @@ -45,6 +45,7 @@ _raise_bulk_write_error, _Run, ) +from pymongo.client_session_shared import _validate_session_write_concern from pymongo.common import ( validate_is_document_type, validate_ok_for_replace, diff --git a/pymongo/asynchronous/client_bulk.py b/pymongo/asynchronous/client_bulk.py index 367fdd492f..2847dcb2c0 100644 --- a/pymongo/asynchronous/client_bulk.py +++ b/pymongo/asynchronous/client_bulk.py @@ -33,10 +33,7 @@ from bson.raw_bson import RawBSONDocument from pymongo import _csot, common from pymongo._telemetry import _generate_op_id_or_none -from pymongo.asynchronous.client_session import ( - AsyncClientSession, - _validate_session_write_concern, -) +from pymongo.asynchronous.client_session import AsyncClientSession from pymongo.asynchronous.collection import AsyncCollection from pymongo.asynchronous.command_cursor import AsyncCommandCursor from pymongo.asynchronous.command_runner import ( @@ -52,6 +49,7 @@ _merge_command, _throw_client_bulk_write_exception, ) +from pymongo.client_session_shared import _validate_session_write_concern from pymongo.common import ( validate_is_document_type, validate_ok_for_replace, diff --git a/pymongo/asynchronous/client_session.py b/pymongo/asynchronous/client_session.py index 72b1ac100e..acadb33ce3 100644 --- a/pymongo/asynchronous/client_session.py +++ b/pymongo/asynchronous/client_session.py @@ -136,10 +136,8 @@ from __future__ import annotations import asyncio -import collections import random import time -import uuid from collections.abc import Awaitable, Mapping, MutableMapping from collections.abc import Mapping as _Mapping from contextlib import AbstractAsyncContextManager @@ -153,22 +151,31 @@ TypeVar, ) -from bson.binary import Binary from bson.int64 import Int64 from bson.timestamp import Timestamp from pymongo import _csot from pymongo.asynchronous.cursor_base import _ConnectionManager +from pymongo.client_session_shared import ( + _BACKOFF_INITIAL, + _BACKOFF_MAX, + _UNKNOWN_COMMIT_ERROR_CODES, + SessionOptions, + TransactionOptions, + _EmptyServerSession, + _make_timeout_error, + _max_time_expired_error, + _reraise_with_unknown_commit, + _TxnState, + _within_time_limit, +) from pymongo.errors import ( ConfigurationError, ConnectionFailure, - ExecutionTimeout, InvalidOperation, - NetworkTimeout, OperationFailure, PyMongoError, WTimeoutError, ) -from pymongo.helpers_shared import _RETRYABLE_ERROR_CODES from pymongo.operations import _WRITES_WITH_CLUSTER_TIME from pymongo.read_concern import ReadConcern from pymongo.read_preferences import ReadPreference, _ServerMode @@ -207,182 +214,6 @@ async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: await self._session.end_session() -class SessionOptions: - """Options for a new :class:`AsyncClientSession`. - - :param causal_consistency: If True, read operations are causally - ordered within the session. Defaults to True when the ``snapshot`` - option is ``False``. - :param default_transaction_options: The default - TransactionOptions to use for transactions started on this session. - :param snapshot: If True, then all reads performed using this - session will read from the same snapshot. This option is incompatible - with ``causal_consistency=True``. Defaults to ``False``. - - .. versionchanged:: 3.12 - Added the ``snapshot`` parameter. - """ - - def __init__( - self, - causal_consistency: Optional[bool] = None, - default_transaction_options: Optional[TransactionOptions] = None, - snapshot: Optional[bool] = False, - ) -> None: - if snapshot: - if causal_consistency: - raise ConfigurationError("snapshot reads do not support causal_consistency=True") - causal_consistency = False - elif causal_consistency is None: - causal_consistency = True - self._causal_consistency = causal_consistency - if default_transaction_options is not None: - if not isinstance(default_transaction_options, TransactionOptions): - raise TypeError( - "default_transaction_options must be an instance of " - f"pymongo.client_session.TransactionOptions, not: {default_transaction_options!r}" - ) - self._default_transaction_options = default_transaction_options - self._snapshot = snapshot - - @property - def causal_consistency(self) -> bool: - """Whether causal consistency is configured.""" - return self._causal_consistency - - @property - def default_transaction_options(self) -> Optional[TransactionOptions]: - """The default TransactionOptions to use for transactions started on - this session. - - .. versionadded:: 3.7 - """ - return self._default_transaction_options - - @property - def snapshot(self) -> Optional[bool]: - """Whether snapshot reads are configured. - - .. versionadded:: 3.12 - """ - return self._snapshot - - -class TransactionOptions: - """Options for :meth:`AsyncClientSession.start_transaction`. - - :param read_concern: The - :class:`~pymongo.read_concern.ReadConcern` to use for this transaction. - If ``None`` (the default) the :attr:`read_preference` of - the :class:`AsyncMongoClient` is used. - :param write_concern: The - :class:`~pymongo.write_concern.WriteConcern` to use for this - transaction. If ``None`` (the default) the :attr:`read_preference` of - the :class:`AsyncMongoClient` is used. - :param read_preference: The read preference to use. If - ``None`` (the default) the :attr:`read_preference` of this - :class:`AsyncMongoClient` is used. See :mod:`~pymongo.read_preferences` - for options. Transactions which read must use - :attr:`~pymongo.read_preferences.ReadPreference.PRIMARY`. - :param max_commit_time_ms: The maximum amount of time to allow a - single commitTransaction command to run. This option is an alias for - maxTimeMS option on the commitTransaction command. If ``None`` (the - default) maxTimeMS is not used. - - .. versionchanged:: 3.9 - Added the ``max_commit_time_ms`` option. - - .. versionadded:: 3.7 - """ - - def __init__( - self, - read_concern: Optional[ReadConcern] = None, - write_concern: Optional[WriteConcern] = None, - read_preference: Optional[_ServerMode] = None, - max_commit_time_ms: Optional[int] = None, - ) -> None: - self._read_concern = read_concern - self._write_concern = write_concern - self._read_preference = read_preference - self._max_commit_time_ms = max_commit_time_ms - if read_concern is not None: - if not isinstance(read_concern, ReadConcern): - raise TypeError( - "read_concern must be an instance of " - f"pymongo.read_concern.ReadConcern, not: {read_concern!r}" - ) - if write_concern is not None: - if not isinstance(write_concern, WriteConcern): - raise TypeError( - "write_concern must be an instance of " - f"pymongo.write_concern.WriteConcern, not: {write_concern!r}" - ) - if not write_concern.acknowledged: - raise ConfigurationError( - f"transactions do not support unacknowledged write concern: {write_concern!r}" - ) - if read_preference is not None: - if not isinstance(read_preference, _ServerMode): - raise TypeError( - f"{read_preference!r} is not valid for read_preference. See " - "pymongo.read_preferences for valid " - "options." - ) - if max_commit_time_ms is not None: - if not isinstance(max_commit_time_ms, int): - raise TypeError( - f"max_commit_time_ms must be an integer or None, not {type(max_commit_time_ms)}" - ) - - @property - def read_concern(self) -> Optional[ReadConcern]: - """This transaction's :class:`~pymongo.read_concern.ReadConcern`.""" - return self._read_concern - - @property - def write_concern(self) -> Optional[WriteConcern]: - """This transaction's :class:`~pymongo.write_concern.WriteConcern`.""" - return self._write_concern - - @property - def read_preference(self) -> Optional[_ServerMode]: - """This transaction's :class:`~pymongo.read_preferences.ReadPreference`.""" - return self._read_preference - - @property - def max_commit_time_ms(self) -> Optional[int]: - """The maxTimeMS to use when running a commitTransaction command. - - .. versionadded:: 3.9 - """ - return self._max_commit_time_ms - - -def _validate_session_write_concern( - session: Optional[AsyncClientSession], write_concern: Optional[WriteConcern] -) -> Optional[AsyncClientSession]: - """Validate that an explicit session is not used with an unack'ed write. - - Returns the session to use for the next operation. - """ - if session: - if write_concern is not None and not write_concern.acknowledged: - # For unacknowledged writes without an explicit session, - # drivers SHOULD NOT use an implicit session. If a driver - # creates an implicit session for unacknowledged writes - # without an explicit session, the driver MUST NOT send the - # session ID. - if session._implicit: - return None - else: - raise ConfigurationError( - "Explicit sessions are incompatible with " - f"unacknowledged write concern: {write_concern!r}" - ) - return session - - class _TransactionContext: """Internal transaction context manager for start_transaction.""" @@ -405,15 +236,6 @@ async def __aexit__( await self.__session.abort_transaction() -class _TxnState: - NONE = 1 - STARTING = 2 - IN_PROGRESS = 3 - COMMITTED = 4 - COMMITTED_EMPTY = 5 - ABORTED = 6 - - class _Transaction: """Internal class to hold transaction information in a AsyncClientSession.""" @@ -476,56 +298,6 @@ def __del__(self) -> None: self.conn_mgr = None -def _reraise_with_unknown_commit(exc: Any) -> NoReturn: - """Re-raise an exception with the UnknownTransactionCommitResult label.""" - exc._add_error_label("UnknownTransactionCommitResult") - raise exc - - -def _max_time_expired_error(exc: PyMongoError) -> bool: - """Return true if exc is a MaxTimeMSExpired error.""" - return isinstance(exc, OperationFailure) and exc.code == 50 - - -# From the transactions spec, all the retryable writes errors plus -# WriteConcernTimeout. -_UNKNOWN_COMMIT_ERROR_CODES: frozenset = _RETRYABLE_ERROR_CODES | frozenset( # type: ignore[type-arg] - [ - 64, # WriteConcernTimeout - 50, # MaxTimeMSExpired - ] -) - -# From the Convenient API for Transactions spec, with_transaction must -# halt retries after 120 seconds. -# This limit is non-configurable and was chosen to be twice the 60 second -# default value of MongoDB's `transactionLifetimeLimitSeconds` parameter. -_WITH_TRANSACTION_RETRY_TIME_LIMIT = 120 -_BACKOFF_MAX = 0.500 # 500ms max backoff -_BACKOFF_INITIAL = 0.005 # 5ms initial backoff - - -def _within_time_limit(start_time: float, backoff: float = 0) -> bool: - """Are we within the with_transaction retry limit?""" - remaining = _csot.remaining() - if remaining is not None and remaining <= 0: - return False - return time.monotonic() + backoff - start_time < _WITH_TRANSACTION_RETRY_TIME_LIMIT - - -def _make_timeout_error(error: BaseException) -> PyMongoError: - """Convert error to a NetworkTimeout or ExecutionTimeout as appropriate.""" - if _csot.remaining() is not None: - timeout_error: PyMongoError = ExecutionTimeout( - str(error), 50, {"ok": 0, "errmsg": str(error), "code": 50} - ) - else: - timeout_error = NetworkTimeout(str(error)) - if isinstance(error, PyMongoError): - timeout_error._error_labels = error._error_labels.copy() - return timeout_error - - _T = TypeVar("_T") if TYPE_CHECKING: @@ -1162,113 +934,3 @@ def _update_read_concern(self, cmd: MutableMapping[str, Any], conn: AsyncConnect def __copy__(self) -> NoReturn: raise TypeError("A AsyncClientSession cannot be copied, create a new session instead") - - -class _EmptyServerSession: - __slots__ = "dirty", "started_retryable_write" - - def __init__(self) -> None: - self.dirty = False - self.started_retryable_write = False - - def mark_dirty(self) -> None: - self.dirty = True - - def inc_transaction_id(self) -> None: - self.started_retryable_write = True - - -class _ServerSession: - def __init__(self, generation: int): - # Ensure id is type 4, regardless of CodecOptions.uuid_representation. - self.session_id = {"id": Binary(uuid.uuid4().bytes, 4)} - self.last_use = time.monotonic() - self._transaction_id = 0 - self.dirty = False - self.generation = generation - - def mark_dirty(self) -> None: - """Mark this session as dirty. - - A server session is marked dirty when a command fails with a network - error. Dirty sessions are later discarded from the server session pool. - """ - self.dirty = True - - def timed_out(self, session_timeout_minutes: Optional[int]) -> bool: - if session_timeout_minutes is None: - return False - - idle_seconds = time.monotonic() - self.last_use - - # Timed out if we have less than a minute to live. - return idle_seconds > (session_timeout_minutes - 1) * 60 - - @property - def transaction_id(self) -> Int64: - """Positive 64-bit integer.""" - return Int64(self._transaction_id) - - def inc_transaction_id(self) -> None: - self._transaction_id += 1 - - -class _ServerSessionPool(collections.deque): # type: ignore[type-arg] - """Pool of _ServerSession objects. - - This class is thread-safe. - """ - - def __init__(self, *args: Any, **kwargs: Any): - super().__init__(*args, **kwargs) - self.generation = 0 - - def reset(self) -> None: - self.generation += 1 - self.clear() - - def pop_all(self) -> list[_ServerSession]: - ids = [] - 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: - # Although the Driver Sessions Spec says we only clear stale sessions - # in return_server_session, PyMongo can't take a lock when returning - # sessions from a __del__ method (like in AsyncCursor.__die), so it can't - # clear stale sessions there. In case many sessions were returned via - # __del__, check for stale sessions here too. - self._clear_stale(session_timeout_minutes) - - # The most recently used sessions are on the left. - 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) -> 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 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 diff --git a/pymongo/asynchronous/mongo_client.py b/pymongo/asynchronous/mongo_client.py index baa281b59a..a4d75e480e 100644 --- a/pymongo/asynchronous/mongo_client.py +++ b/pymongo/asynchronous/mongo_client.py @@ -60,7 +60,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 _SESSION, _EmptyServerSession +from pymongo.asynchronous.client_session import _SESSION from pymongo.asynchronous.command_cursor import AsyncCommandCursor from pymongo.asynchronous.helpers import ( _RetryPolicy, @@ -68,6 +68,7 @@ from pymongo.asynchronous.settings import TopologySettings from pymongo.asynchronous.topology import Topology, _ErrorContext from pymongo.client_options import ClientOptions +from pymongo.client_session_shared import SessionOptions, TransactionOptions, _EmptyServerSession from pymongo.driver_info import DriverInfo from pymongo.errors import ( AutoReconnect, @@ -133,11 +134,12 @@ from bson.objectid import ObjectId from pymongo.asynchronous.bulk import _AsyncBulk - from pymongo.asynchronous.client_session import AsyncClientSession, _ServerSession + from pymongo.asynchronous.client_session import AsyncClientSession from pymongo.asynchronous.cursor_base import _ConnectionManager from pymongo.asynchronous.encryption import _Encrypter from pymongo.asynchronous.pool import AsyncConnection, _PoolCheckout from pymongo.asynchronous.server import Server + from pymongo.client_session_shared import _ServerSession from pymongo.read_concern import ReadConcern from pymongo.response import Response from pymongo.server_selectors import Selection @@ -1388,13 +1390,13 @@ def _close_cursor_soon( def _start_session(self, implicit: bool, **kwargs: Any) -> AsyncClientSession: server_session = _EmptyServerSession() - opts = client_session.SessionOptions(**kwargs) + opts = SessionOptions(**kwargs) return client_session.AsyncClientSession(self, server_session, opts, implicit) def start_session( self, causal_consistency: Optional[bool] = None, - default_transaction_options: Optional[client_session.TransactionOptions] = None, + default_transaction_options: Optional[TransactionOptions] = None, snapshot: Optional[bool] = False, ) -> client_session.AsyncClientSession: """Start a logical session. diff --git a/pymongo/asynchronous/pool.py b/pymongo/asynchronous/pool.py index eb9cb215eb..506ed5676e 100644 --- a/pymongo/asynchronous/pool.py +++ b/pymongo/asynchronous/pool.py @@ -32,9 +32,9 @@ from bson import DEFAULT_CODEC_OPTIONS from pymongo import _csot, helpers_shared from pymongo._telemetry import _CmapTelemetry -from pymongo.asynchronous.client_session import _validate_session_write_concern from pymongo.asynchronous.command_runner import run_command from pymongo.asynchronous.helpers import _handle_reauth +from pymongo.client_session_shared import _validate_session_write_concern from pymongo.common import ( MAX_BSON_SIZE, MAX_MESSAGE_SIZE, diff --git a/pymongo/asynchronous/topology.py b/pymongo/asynchronous/topology.py index 72481296ff..ffa0626840 100644 --- a/pymongo/asynchronous/topology.py +++ b/pymongo/asynchronous/topology.py @@ -34,10 +34,10 @@ _ServerSelectionTelemetry, log_server_selection_succeeded, ) -from pymongo.asynchronous.client_session import _ServerSession, _ServerSessionPool from pymongo.asynchronous.monitor import MonitorBase, SrvMonitor from pymongo.asynchronous.pool import Pool from pymongo.asynchronous.server import Server +from pymongo.client_session_shared import _ServerSession, _ServerSessionPool from pymongo.errors import ( ConnectionFailure, InvalidOperation, diff --git a/pymongo/client_session_shared.py b/pymongo/client_session_shared.py new file mode 100644 index 0000000000..467d069355 --- /dev/null +++ b/pymongo/client_session_shared.py @@ -0,0 +1,395 @@ +# Copyright 2017-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 +# +# https://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. + +"""Internal helpers for logical sessions, shared between the asynchronous and synchronous APIs.""" + +from __future__ import annotations + +import collections +import time +import uuid +from typing import ( + TYPE_CHECKING, + Any, + NoReturn, + Optional, + TypeVar, +) + +from bson.binary import Binary +from bson.int64 import Int64 +from pymongo import _csot +from pymongo.errors import ( + ConfigurationError, + ExecutionTimeout, + NetworkTimeout, + OperationFailure, + PyMongoError, +) +from pymongo.helpers_shared import _RETRYABLE_ERROR_CODES +from pymongo.read_concern import ReadConcern +from pymongo.read_preferences import _ServerMode +from pymongo.write_concern import WriteConcern + +if TYPE_CHECKING: + from pymongo.typings import _AgnosticClientSession + +_ClientSessionT = TypeVar("_ClientSessionT", bound="_AgnosticClientSession") + + +class SessionOptions: + """Options for a new :class:`~pymongo.asynchronous.client_session.AsyncClientSession` + or :class:`~pymongo.client_session.ClientSession`. + + :param causal_consistency: If True, read operations are causally + ordered within the session. Defaults to True when the ``snapshot`` + option is ``False``. + :param default_transaction_options: The default + TransactionOptions to use for transactions started on this session. + :param snapshot: If True, then all reads performed using this + session will read from the same snapshot. This option is incompatible + with ``causal_consistency=True``. Defaults to ``False``. + + .. versionchanged:: 3.12 + Added the ``snapshot`` parameter. + """ + + def __init__( + self, + causal_consistency: Optional[bool] = None, + default_transaction_options: Optional[TransactionOptions] = None, + snapshot: Optional[bool] = False, + ) -> None: + if snapshot: + if causal_consistency: + raise ConfigurationError("snapshot reads do not support causal_consistency=True") + causal_consistency = False + elif causal_consistency is None: + causal_consistency = True + self._causal_consistency = causal_consistency + if default_transaction_options is not None: + if not isinstance(default_transaction_options, TransactionOptions): + raise TypeError( + "default_transaction_options must be an instance of " + f"pymongo.client_session.TransactionOptions, not: {default_transaction_options!r}" + ) + self._default_transaction_options = default_transaction_options + self._snapshot = snapshot + + @property + def causal_consistency(self) -> bool: + """Whether causal consistency is configured.""" + return self._causal_consistency + + @property + def default_transaction_options(self) -> Optional[TransactionOptions]: + """The default TransactionOptions to use for transactions started on + this session. + + .. versionadded:: 3.7 + """ + return self._default_transaction_options + + @property + def snapshot(self) -> Optional[bool]: + """Whether snapshot reads are configured. + + .. versionadded:: 3.12 + """ + return self._snapshot + + +class TransactionOptions: + """Options for :meth:`~pymongo.asynchronous.client_session.AsyncClientSession.start_transaction` + or :meth:`~pymongo.client_session.ClientSession.start_transaction`. + + :param read_concern: The + :class:`~pymongo.read_concern.ReadConcern` to use for this transaction. + If ``None`` (the default) the :attr:`read_preference` of + the client is used. + :param write_concern: The + :class:`~pymongo.write_concern.WriteConcern` to use for this + transaction. If ``None`` (the default) the :attr:`read_preference` of + the client is used. + :param read_preference: The read preference to use. If + ``None`` (the default) the :attr:`read_preference` of this + client is used. See :mod:`~pymongo.read_preferences` + for options. Transactions which read must use + :attr:`~pymongo.read_preferences.ReadPreference.PRIMARY`. + :param max_commit_time_ms: The maximum amount of time to allow a + single commitTransaction command to run. This option is an alias for + maxTimeMS option on the commitTransaction command. If ``None`` (the + default) maxTimeMS is not used. + + .. versionchanged:: 3.9 + Added the ``max_commit_time_ms`` option. + + .. versionadded:: 3.7 + """ + + def __init__( + self, + read_concern: Optional[ReadConcern] = None, + write_concern: Optional[WriteConcern] = None, + read_preference: Optional[_ServerMode] = None, + max_commit_time_ms: Optional[int] = None, + ) -> None: + self._read_concern = read_concern + self._write_concern = write_concern + self._read_preference = read_preference + self._max_commit_time_ms = max_commit_time_ms + if read_concern is not None: + if not isinstance(read_concern, ReadConcern): + raise TypeError( + "read_concern must be an instance of " + f"pymongo.read_concern.ReadConcern, not: {read_concern!r}" + ) + if write_concern is not None: + if not isinstance(write_concern, WriteConcern): + raise TypeError( + "write_concern must be an instance of " + f"pymongo.write_concern.WriteConcern, not: {write_concern!r}" + ) + if not write_concern.acknowledged: + raise ConfigurationError( + f"transactions do not support unacknowledged write concern: {write_concern!r}" + ) + if read_preference is not None: + if not isinstance(read_preference, _ServerMode): + raise TypeError( + f"{read_preference!r} is not valid for read_preference. See " + "pymongo.read_preferences for valid " + "options." + ) + if max_commit_time_ms is not None: + if not isinstance(max_commit_time_ms, int): + raise TypeError( + f"max_commit_time_ms must be an integer or None, not {type(max_commit_time_ms)}" + ) + + @property + def read_concern(self) -> Optional[ReadConcern]: + """This transaction's :class:`~pymongo.read_concern.ReadConcern`.""" + return self._read_concern + + @property + def write_concern(self) -> Optional[WriteConcern]: + """This transaction's :class:`~pymongo.write_concern.WriteConcern`.""" + return self._write_concern + + @property + def read_preference(self) -> Optional[_ServerMode]: + """This transaction's :class:`~pymongo.read_preferences.ReadPreference`.""" + return self._read_preference + + @property + def max_commit_time_ms(self) -> Optional[int]: + """The maxTimeMS to use when running a commitTransaction command. + + .. versionadded:: 3.9 + """ + return self._max_commit_time_ms + + +def _validate_session_write_concern( + session: Optional[_ClientSessionT], write_concern: Optional[WriteConcern] +) -> Optional[_ClientSessionT]: + """Validate that an explicit session is not used with an unack'ed write. + + Returns the session to use for the next operation. + """ + if session: + if write_concern is not None and not write_concern.acknowledged: + # For unacknowledged writes without an explicit session, + # drivers SHOULD NOT use an implicit session. If a driver + # creates an implicit session for unacknowledged writes + # without an explicit session, the driver MUST NOT send the + # session ID. + if session._implicit: + return None + else: + raise ConfigurationError( + "Explicit sessions are incompatible with " + f"unacknowledged write concern: {write_concern!r}" + ) + return session + + +class _TxnState: + NONE = 1 + STARTING = 2 + IN_PROGRESS = 3 + COMMITTED = 4 + COMMITTED_EMPTY = 5 + ABORTED = 6 + + +def _reraise_with_unknown_commit(exc: Any) -> NoReturn: + """Re-raise an exception with the UnknownTransactionCommitResult label.""" + exc._add_error_label("UnknownTransactionCommitResult") + raise exc + + +def _max_time_expired_error(exc: PyMongoError) -> bool: + """Return true if exc is a MaxTimeMSExpired error.""" + return isinstance(exc, OperationFailure) and exc.code == 50 + + +# From the transactions spec, all the retryable writes errors plus +# WriteConcernTimeout. +_UNKNOWN_COMMIT_ERROR_CODES: frozenset = _RETRYABLE_ERROR_CODES | frozenset( # type: ignore[type-arg] + [ + 64, # WriteConcernTimeout + 50, # MaxTimeMSExpired + ] +) + +# From the Convenient API for Transactions spec, with_transaction must +# halt retries after 120 seconds. +# This limit is non-configurable and was chosen to be twice the 60 second +# default value of MongoDB's `transactionLifetimeLimitSeconds` parameter. +_WITH_TRANSACTION_RETRY_TIME_LIMIT = 120 +_BACKOFF_MAX = 0.500 # 500ms max backoff +_BACKOFF_INITIAL = 0.005 # 5ms initial backoff + + +def _within_time_limit(start_time: float, backoff: float = 0) -> bool: + """Are we within the with_transaction retry limit?""" + remaining = _csot.remaining() + if remaining is not None and remaining <= 0: + return False + return time.monotonic() + backoff - start_time < _WITH_TRANSACTION_RETRY_TIME_LIMIT + + +def _make_timeout_error(error: BaseException) -> PyMongoError: + """Convert error to a NetworkTimeout or ExecutionTimeout as appropriate.""" + if _csot.remaining() is not None: + timeout_error: PyMongoError = ExecutionTimeout( + str(error), 50, {"ok": 0, "errmsg": str(error), "code": 50} + ) + else: + timeout_error = NetworkTimeout(str(error)) + if isinstance(error, PyMongoError): + timeout_error._error_labels = error._error_labels.copy() + return timeout_error + + +class _EmptyServerSession: + __slots__ = "dirty", "started_retryable_write" + + def __init__(self) -> None: + self.dirty = False + self.started_retryable_write = False + + def mark_dirty(self) -> None: + self.dirty = True + + def inc_transaction_id(self) -> None: + self.started_retryable_write = True + + +class _ServerSession: + def __init__(self, generation: int): + # Ensure id is type 4, regardless of CodecOptions.uuid_representation. + self.session_id = {"id": Binary(uuid.uuid4().bytes, 4)} + self.last_use = time.monotonic() + self._transaction_id = 0 + self.dirty = False + self.generation = generation + + def mark_dirty(self) -> None: + """Mark this session as dirty. + + A server session is marked dirty when a command fails with a network + error. Dirty sessions are later discarded from the server session pool. + """ + self.dirty = True + + def timed_out(self, session_timeout_minutes: Optional[int]) -> bool: + if session_timeout_minutes is None: + return False + + idle_seconds = time.monotonic() - self.last_use + + # Timed out if we have less than a minute to live. + return idle_seconds > (session_timeout_minutes - 1) * 60 + + @property + def transaction_id(self) -> Int64: + """Positive 64-bit integer.""" + return Int64(self._transaction_id) + + def inc_transaction_id(self) -> None: + self._transaction_id += 1 + + +class _ServerSessionPool(collections.deque): # type: ignore[type-arg] + """Pool of _ServerSession objects. + + This class is thread-safe. + """ + + def __init__(self, *args: Any, **kwargs: Any): + super().__init__(*args, **kwargs) + self.generation = 0 + + def reset(self) -> None: + self.generation += 1 + self.clear() + + def pop_all(self) -> list[_ServerSession]: + ids = [] + 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: + # Although the Driver Sessions Spec says we only clear stale sessions + # in return_server_session, PyMongo can't take a lock when returning + # sessions from a __del__ method (like in Cursor.__die), so it can't + # clear stale sessions there. In case many sessions were returned via + # __del__, check for stale sessions here too. + self._clear_stale(session_timeout_minutes) + + # The most recently used sessions are on the left. + 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) -> 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 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 diff --git a/pymongo/synchronous/bulk.py b/pymongo/synchronous/bulk.py index 79d64c07c2..be12a72127 100644 --- a/pymongo/synchronous/bulk.py +++ b/pymongo/synchronous/bulk.py @@ -40,6 +40,7 @@ _raise_bulk_write_error, _Run, ) +from pymongo.client_session_shared import _validate_session_write_concern from pymongo.common import ( validate_is_document_type, validate_ok_for_replace, @@ -59,7 +60,7 @@ _EncryptedBulkWriteContext, ) from pymongo.read_preferences import ReadPreference -from pymongo.synchronous.client_session import ClientSession, _validate_session_write_concern +from pymongo.synchronous.client_session import ClientSession from pymongo.synchronous.command_runner import ( run_bulk_write_command, ) diff --git a/pymongo/synchronous/client_bulk.py b/pymongo/synchronous/client_bulk.py index 3dca2f7234..5e5d1a76e8 100644 --- a/pymongo/synchronous/client_bulk.py +++ b/pymongo/synchronous/client_bulk.py @@ -33,10 +33,7 @@ from bson.raw_bson import RawBSONDocument from pymongo import _csot, common from pymongo._telemetry import _generate_op_id_or_none -from pymongo.synchronous.client_session import ( - ClientSession, - _validate_session_write_concern, -) +from pymongo.synchronous.client_session import ClientSession from pymongo.synchronous.collection import Collection from pymongo.synchronous.command_cursor import CommandCursor from pymongo.synchronous.command_runner import ( @@ -52,6 +49,7 @@ _merge_command, _throw_client_bulk_write_exception, ) +from pymongo.client_session_shared import _validate_session_write_concern from pymongo.common import ( validate_is_document_type, validate_ok_for_replace, diff --git a/pymongo/synchronous/client_session.py b/pymongo/synchronous/client_session.py index 0774c182de..30cb17e14f 100644 --- a/pymongo/synchronous/client_session.py +++ b/pymongo/synchronous/client_session.py @@ -135,10 +135,8 @@ from __future__ import annotations -import collections import random import time -import uuid from collections.abc import Mapping, MutableMapping from collections.abc import Mapping as _Mapping from contextlib import AbstractContextManager @@ -152,21 +150,30 @@ TypeVar, ) -from bson.binary import Binary from bson.int64 import Int64 from bson.timestamp import Timestamp from pymongo import _csot +from pymongo.client_session_shared import ( + _BACKOFF_INITIAL, + _BACKOFF_MAX, + _UNKNOWN_COMMIT_ERROR_CODES, + SessionOptions, + TransactionOptions, + _EmptyServerSession, + _make_timeout_error, + _max_time_expired_error, + _reraise_with_unknown_commit, + _TxnState, + _within_time_limit, +) from pymongo.errors import ( ConfigurationError, ConnectionFailure, - ExecutionTimeout, InvalidOperation, - NetworkTimeout, OperationFailure, PyMongoError, WTimeoutError, ) -from pymongo.helpers_shared import _RETRYABLE_ERROR_CODES from pymongo.operations import _WRITES_WITH_CLUSTER_TIME from pymongo.read_concern import ReadConcern from pymongo.read_preferences import ReadPreference, _ServerMode @@ -206,182 +213,6 @@ def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: self._session.end_session() -class SessionOptions: - """Options for a new :class:`ClientSession`. - - :param causal_consistency: If True, read operations are causally - ordered within the session. Defaults to True when the ``snapshot`` - option is ``False``. - :param default_transaction_options: The default - TransactionOptions to use for transactions started on this session. - :param snapshot: If True, then all reads performed using this - session will read from the same snapshot. This option is incompatible - with ``causal_consistency=True``. Defaults to ``False``. - - .. versionchanged:: 3.12 - Added the ``snapshot`` parameter. - """ - - def __init__( - self, - causal_consistency: Optional[bool] = None, - default_transaction_options: Optional[TransactionOptions] = None, - snapshot: Optional[bool] = False, - ) -> None: - if snapshot: - if causal_consistency: - raise ConfigurationError("snapshot reads do not support causal_consistency=True") - causal_consistency = False - elif causal_consistency is None: - causal_consistency = True - self._causal_consistency = causal_consistency - if default_transaction_options is not None: - if not isinstance(default_transaction_options, TransactionOptions): - raise TypeError( - "default_transaction_options must be an instance of " - f"pymongo.client_session.TransactionOptions, not: {default_transaction_options!r}" - ) - self._default_transaction_options = default_transaction_options - self._snapshot = snapshot - - @property - def causal_consistency(self) -> bool: - """Whether causal consistency is configured.""" - return self._causal_consistency - - @property - def default_transaction_options(self) -> Optional[TransactionOptions]: - """The default TransactionOptions to use for transactions started on - this session. - - .. versionadded:: 3.7 - """ - return self._default_transaction_options - - @property - def snapshot(self) -> Optional[bool]: - """Whether snapshot reads are configured. - - .. versionadded:: 3.12 - """ - return self._snapshot - - -class TransactionOptions: - """Options for :meth:`ClientSession.start_transaction`. - - :param read_concern: The - :class:`~pymongo.read_concern.ReadConcern` to use for this transaction. - If ``None`` (the default) the :attr:`read_preference` of - the :class:`MongoClient` is used. - :param write_concern: The - :class:`~pymongo.write_concern.WriteConcern` to use for this - transaction. If ``None`` (the default) the :attr:`read_preference` of - the :class:`MongoClient` is used. - :param read_preference: The read preference to use. If - ``None`` (the default) the :attr:`read_preference` of this - :class:`MongoClient` is used. See :mod:`~pymongo.read_preferences` - for options. Transactions which read must use - :attr:`~pymongo.read_preferences.ReadPreference.PRIMARY`. - :param max_commit_time_ms: The maximum amount of time to allow a - single commitTransaction command to run. This option is an alias for - maxTimeMS option on the commitTransaction command. If ``None`` (the - default) maxTimeMS is not used. - - .. versionchanged:: 3.9 - Added the ``max_commit_time_ms`` option. - - .. versionadded:: 3.7 - """ - - def __init__( - self, - read_concern: Optional[ReadConcern] = None, - write_concern: Optional[WriteConcern] = None, - read_preference: Optional[_ServerMode] = None, - max_commit_time_ms: Optional[int] = None, - ) -> None: - self._read_concern = read_concern - self._write_concern = write_concern - self._read_preference = read_preference - self._max_commit_time_ms = max_commit_time_ms - if read_concern is not None: - if not isinstance(read_concern, ReadConcern): - raise TypeError( - "read_concern must be an instance of " - f"pymongo.read_concern.ReadConcern, not: {read_concern!r}" - ) - if write_concern is not None: - if not isinstance(write_concern, WriteConcern): - raise TypeError( - "write_concern must be an instance of " - f"pymongo.write_concern.WriteConcern, not: {write_concern!r}" - ) - if not write_concern.acknowledged: - raise ConfigurationError( - f"transactions do not support unacknowledged write concern: {write_concern!r}" - ) - if read_preference is not None: - if not isinstance(read_preference, _ServerMode): - raise TypeError( - f"{read_preference!r} is not valid for read_preference. See " - "pymongo.read_preferences for valid " - "options." - ) - if max_commit_time_ms is not None: - if not isinstance(max_commit_time_ms, int): - raise TypeError( - f"max_commit_time_ms must be an integer or None, not {type(max_commit_time_ms)}" - ) - - @property - def read_concern(self) -> Optional[ReadConcern]: - """This transaction's :class:`~pymongo.read_concern.ReadConcern`.""" - return self._read_concern - - @property - def write_concern(self) -> Optional[WriteConcern]: - """This transaction's :class:`~pymongo.write_concern.WriteConcern`.""" - return self._write_concern - - @property - def read_preference(self) -> Optional[_ServerMode]: - """This transaction's :class:`~pymongo.read_preferences.ReadPreference`.""" - return self._read_preference - - @property - def max_commit_time_ms(self) -> Optional[int]: - """The maxTimeMS to use when running a commitTransaction command. - - .. versionadded:: 3.9 - """ - return self._max_commit_time_ms - - -def _validate_session_write_concern( - session: Optional[ClientSession], write_concern: Optional[WriteConcern] -) -> Optional[ClientSession]: - """Validate that an explicit session is not used with an unack'ed write. - - Returns the session to use for the next operation. - """ - if session: - if write_concern is not None and not write_concern.acknowledged: - # For unacknowledged writes without an explicit session, - # drivers SHOULD NOT use an implicit session. If a driver - # creates an implicit session for unacknowledged writes - # without an explicit session, the driver MUST NOT send the - # session ID. - if session._implicit: - return None - else: - raise ConfigurationError( - "Explicit sessions are incompatible with " - f"unacknowledged write concern: {write_concern!r}" - ) - return session - - class _TransactionContext: """Internal transaction context manager for start_transaction.""" @@ -404,15 +235,6 @@ def __exit__( self.__session.abort_transaction() -class _TxnState: - NONE = 1 - STARTING = 2 - IN_PROGRESS = 3 - COMMITTED = 4 - COMMITTED_EMPTY = 5 - ABORTED = 6 - - class _Transaction: """Internal class to hold transaction information in a ClientSession.""" @@ -475,56 +297,6 @@ def __del__(self) -> None: self.conn_mgr = None -def _reraise_with_unknown_commit(exc: Any) -> NoReturn: - """Re-raise an exception with the UnknownTransactionCommitResult label.""" - exc._add_error_label("UnknownTransactionCommitResult") - raise exc - - -def _max_time_expired_error(exc: PyMongoError) -> bool: - """Return true if exc is a MaxTimeMSExpired error.""" - return isinstance(exc, OperationFailure) and exc.code == 50 - - -# From the transactions spec, all the retryable writes errors plus -# WriteConcernTimeout. -_UNKNOWN_COMMIT_ERROR_CODES: frozenset = _RETRYABLE_ERROR_CODES | frozenset( # type: ignore[type-arg] - [ - 64, # WriteConcernTimeout - 50, # MaxTimeMSExpired - ] -) - -# From the Convenient API for Transactions spec, with_transaction must -# halt retries after 120 seconds. -# This limit is non-configurable and was chosen to be twice the 60 second -# default value of MongoDB's `transactionLifetimeLimitSeconds` parameter. -_WITH_TRANSACTION_RETRY_TIME_LIMIT = 120 -_BACKOFF_MAX = 0.500 # 500ms max backoff -_BACKOFF_INITIAL = 0.005 # 5ms initial backoff - - -def _within_time_limit(start_time: float, backoff: float = 0) -> bool: - """Are we within the with_transaction retry limit?""" - remaining = _csot.remaining() - if remaining is not None and remaining <= 0: - return False - return time.monotonic() + backoff - start_time < _WITH_TRANSACTION_RETRY_TIME_LIMIT - - -def _make_timeout_error(error: BaseException) -> PyMongoError: - """Convert error to a NetworkTimeout or ExecutionTimeout as appropriate.""" - if _csot.remaining() is not None: - timeout_error: PyMongoError = ExecutionTimeout( - str(error), 50, {"ok": 0, "errmsg": str(error), "code": 50} - ) - else: - timeout_error = NetworkTimeout(str(error)) - if isinstance(error, PyMongoError): - timeout_error._error_labels = error._error_labels.copy() - return timeout_error - - _T = TypeVar("_T") if TYPE_CHECKING: @@ -1159,113 +931,3 @@ def _update_read_concern(self, cmd: MutableMapping[str, Any], conn: Connection) def __copy__(self) -> NoReturn: raise TypeError("A ClientSession cannot be copied, create a new session instead") - - -class _EmptyServerSession: - __slots__ = "dirty", "started_retryable_write" - - def __init__(self) -> None: - self.dirty = False - self.started_retryable_write = False - - def mark_dirty(self) -> None: - self.dirty = True - - def inc_transaction_id(self) -> None: - self.started_retryable_write = True - - -class _ServerSession: - def __init__(self, generation: int): - # Ensure id is type 4, regardless of CodecOptions.uuid_representation. - self.session_id = {"id": Binary(uuid.uuid4().bytes, 4)} - self.last_use = time.monotonic() - self._transaction_id = 0 - self.dirty = False - self.generation = generation - - def mark_dirty(self) -> None: - """Mark this session as dirty. - - A server session is marked dirty when a command fails with a network - error. Dirty sessions are later discarded from the server session pool. - """ - self.dirty = True - - def timed_out(self, session_timeout_minutes: Optional[int]) -> bool: - if session_timeout_minutes is None: - return False - - idle_seconds = time.monotonic() - self.last_use - - # Timed out if we have less than a minute to live. - return idle_seconds > (session_timeout_minutes - 1) * 60 - - @property - def transaction_id(self) -> Int64: - """Positive 64-bit integer.""" - return Int64(self._transaction_id) - - def inc_transaction_id(self) -> None: - self._transaction_id += 1 - - -class _ServerSessionPool(collections.deque): # type: ignore[type-arg] - """Pool of _ServerSession objects. - - This class is thread-safe. - """ - - def __init__(self, *args: Any, **kwargs: Any): - super().__init__(*args, **kwargs) - self.generation = 0 - - def reset(self) -> None: - self.generation += 1 - self.clear() - - def pop_all(self) -> list[_ServerSession]: - ids = [] - 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: - # Although the Driver Sessions Spec says we only clear stale sessions - # in return_server_session, PyMongo can't take a lock when returning - # sessions from a __del__ method (like in Cursor.__die), so it can't - # clear stale sessions there. In case many sessions were returned via - # __del__, check for stale sessions here too. - self._clear_stale(session_timeout_minutes) - - # The most recently used sessions are on the left. - 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) -> 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 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 diff --git a/pymongo/synchronous/mongo_client.py b/pymongo/synchronous/mongo_client.py index c6b03201b6..f053ec2a6c 100644 --- a/pymongo/synchronous/mongo_client.py +++ b/pymongo/synchronous/mongo_client.py @@ -58,6 +58,7 @@ from pymongo import _csot, _op_id, common, helpers_shared, periodic_executor from pymongo._telemetry import _generate_op_id_or_none, log_command_retry from pymongo.client_options import ClientOptions +from pymongo.client_session_shared import SessionOptions, TransactionOptions, _EmptyServerSession from pymongo.driver_info import DriverInfo from pymongo.errors import ( AutoReconnect, @@ -101,7 +102,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 _SESSION, _EmptyServerSession +from pymongo.synchronous.client_session import _SESSION from pymongo.synchronous.command_cursor import CommandCursor from pymongo.synchronous.helpers import ( _RetryPolicy, @@ -132,11 +133,12 @@ from types import TracebackType from bson.objectid import ObjectId + from pymongo.client_session_shared import _ServerSession from pymongo.read_concern import ReadConcern from pymongo.response import Response from pymongo.server_selectors import Selection from pymongo.synchronous.bulk import _Bulk - from pymongo.synchronous.client_session import ClientSession, _ServerSession + from pymongo.synchronous.client_session import ClientSession from pymongo.synchronous.cursor_base import _ConnectionManager from pymongo.synchronous.encryption import _Encrypter from pymongo.synchronous.pool import Connection, _PoolCheckout @@ -1389,13 +1391,13 @@ def _close_cursor_soon( def _start_session(self, implicit: bool, **kwargs: Any) -> ClientSession: server_session = _EmptyServerSession() - opts = client_session.SessionOptions(**kwargs) + opts = SessionOptions(**kwargs) return client_session.ClientSession(self, server_session, opts, implicit) def start_session( self, causal_consistency: Optional[bool] = None, - default_transaction_options: Optional[client_session.TransactionOptions] = None, + default_transaction_options: Optional[TransactionOptions] = None, snapshot: Optional[bool] = False, ) -> client_session.ClientSession: """Start a logical session. diff --git a/pymongo/synchronous/pool.py b/pymongo/synchronous/pool.py index 64ff4210a5..958cb709cc 100644 --- a/pymongo/synchronous/pool.py +++ b/pymongo/synchronous/pool.py @@ -32,6 +32,7 @@ from bson import DEFAULT_CODEC_OPTIONS from pymongo import _csot, helpers_shared from pymongo._telemetry import _CmapTelemetry +from pymongo.client_session_shared import _validate_session_write_concern from pymongo.common import ( MAX_BSON_SIZE, MAX_MESSAGE_SIZE, @@ -78,7 +79,6 @@ from pymongo.server_type import SERVER_TYPE from pymongo.socket_checker import SocketChecker from pymongo.ssl_support import SSL_EOF_ERRORS -from pymongo.synchronous.client_session import _validate_session_write_concern from pymongo.synchronous.command_runner import run_command from pymongo.synchronous.helpers import _handle_reauth diff --git a/pymongo/synchronous/topology.py b/pymongo/synchronous/topology.py index cd451218e1..92c1ba6b87 100644 --- a/pymongo/synchronous/topology.py +++ b/pymongo/synchronous/topology.py @@ -34,6 +34,7 @@ _ServerSelectionTelemetry, log_server_selection_succeeded, ) +from pymongo.client_session_shared import _ServerSession, _ServerSessionPool from pymongo.errors import ( ConnectionFailure, InvalidOperation, @@ -61,7 +62,6 @@ secondary_server_selector, writable_server_selector, ) -from pymongo.synchronous.client_session import _ServerSession, _ServerSessionPool from pymongo.synchronous.monitor import MonitorBase, SrvMonitor from pymongo.synchronous.pool import Pool from pymongo.synchronous.server import Server diff --git a/test/asynchronous/test_transactions.py b/test/asynchronous/test_transactions.py index 186dc4fee2..4898d82d97 100644 --- a/test/asynchronous/test_transactions.py +++ b/test/asynchronous/test_transactions.py @@ -35,8 +35,7 @@ from bson import encode from bson.raw_bson import RawBSONDocument -from pymongo import WriteConcern, _csot -from pymongo.asynchronous import client_session +from pymongo import WriteConcern, _csot, client_session_shared from pymongo.asynchronous.client_session import TransactionOptions from pymongo.asynchronous.command_cursor import AsyncCommandCursor from pymongo.asynchronous.cursor import AsyncCursor @@ -402,18 +401,18 @@ async def test_transaction_pool_cleared_error_labelled_transient(self): class PatchSessionTimeout: - """Patches the client_session's with_transaction timeout for testing.""" + """Patches the client_session_shared's with_transaction timeout for testing.""" def __init__(self, mock_timeout): - self.real_timeout = client_session._WITH_TRANSACTION_RETRY_TIME_LIMIT + self.real_timeout = client_session_shared._WITH_TRANSACTION_RETRY_TIME_LIMIT self.mock_timeout = mock_timeout def __enter__(self): - client_session._WITH_TRANSACTION_RETRY_TIME_LIMIT = self.mock_timeout + client_session_shared._WITH_TRANSACTION_RETRY_TIME_LIMIT = self.mock_timeout return self def __exit__(self, exc_type, exc_val, exc_tb): - client_session._WITH_TRANSACTION_RETRY_TIME_LIMIT = self.real_timeout + client_session_shared._WITH_TRANSACTION_RETRY_TIME_LIMIT = self.real_timeout class TestTransactionsConvenientAPI(AsyncTransactionsBase): diff --git a/test/asynchronous/unified_format.py b/test/asynchronous/unified_format.py index d3c1d14d57..5d16bc5918 100644 --- a/test/asynchronous/unified_format.py +++ b/test/asynchronous/unified_format.py @@ -45,12 +45,13 @@ from gridfs.errors import CorruptGridFile from pymongo import ASCENDING, AsyncMongoClient, CursorType, _csot from pymongo.asynchronous.change_stream import AsyncChangeStream -from pymongo.asynchronous.client_session import AsyncClientSession, TransactionOptions, _TxnState +from pymongo.asynchronous.client_session import AsyncClientSession, TransactionOptions from pymongo.asynchronous.collection import AsyncCollection from pymongo.asynchronous.command_cursor import AsyncCommandCursor from pymongo.asynchronous.database import AsyncDatabase from pymongo.asynchronous.encryption import AsyncClientEncryption from pymongo.asynchronous.helpers import anext +from pymongo.client_session_shared import _TxnState from pymongo.driver_info import DriverInfo from pymongo.encryption_options import _HAVE_PYMONGOCRYPT, AutoEncryptionOpts from pymongo.errors import ( diff --git a/test/test_transactions.py b/test/test_transactions.py index 76163c3c12..1b68508ec8 100644 --- a/test/test_transactions.py +++ b/test/test_transactions.py @@ -35,7 +35,7 @@ from bson import encode from bson.raw_bson import RawBSONDocument -from pymongo import WriteConcern, _csot +from pymongo import WriteConcern, _csot, client_session_shared from pymongo.errors import ( AutoReconnect, CollectionInvalid, @@ -49,7 +49,6 @@ from pymongo.operations import IndexModel, InsertOne from pymongo.read_concern import ReadConcern from pymongo.read_preferences import ReadPreference -from pymongo.synchronous import client_session from pymongo.synchronous.client_session import TransactionOptions from pymongo.synchronous.command_cursor import CommandCursor from pymongo.synchronous.cursor import Cursor @@ -394,18 +393,18 @@ def test_transaction_pool_cleared_error_labelled_transient(self): class PatchSessionTimeout: - """Patches the client_session's with_transaction timeout for testing.""" + """Patches the client_session_shared's with_transaction timeout for testing.""" def __init__(self, mock_timeout): - self.real_timeout = client_session._WITH_TRANSACTION_RETRY_TIME_LIMIT + self.real_timeout = client_session_shared._WITH_TRANSACTION_RETRY_TIME_LIMIT self.mock_timeout = mock_timeout def __enter__(self): - client_session._WITH_TRANSACTION_RETRY_TIME_LIMIT = self.mock_timeout + client_session_shared._WITH_TRANSACTION_RETRY_TIME_LIMIT = self.mock_timeout return self def __exit__(self, exc_type, exc_val, exc_tb): - client_session._WITH_TRANSACTION_RETRY_TIME_LIMIT = self.real_timeout + client_session_shared._WITH_TRANSACTION_RETRY_TIME_LIMIT = self.real_timeout class TestTransactionsConvenientAPI(TransactionsBase): diff --git a/test/unified_format.py b/test/unified_format.py index acedd3a5ef..56ea563e6d 100644 --- a/test/unified_format.py +++ b/test/unified_format.py @@ -44,6 +44,7 @@ from gridfs import GridFSBucket, GridOut, NoFile from gridfs.errors import CorruptGridFile from pymongo import ASCENDING, CursorType, MongoClient, _csot +from pymongo.client_session_shared import _TxnState from pymongo.driver_info import DriverInfo from pymongo.encryption_options import _HAVE_PYMONGOCRYPT, AutoEncryptionOpts from pymongo.errors import ( @@ -70,7 +71,7 @@ from pymongo.server_selectors import Selection, writable_server_selector from pymongo.server_type import SERVER_TYPE from pymongo.synchronous.change_stream import ChangeStream -from pymongo.synchronous.client_session import ClientSession, TransactionOptions, _TxnState +from pymongo.synchronous.client_session import ClientSession, TransactionOptions from pymongo.synchronous.collection import Collection from pymongo.synchronous.command_cursor import CommandCursor from pymongo.synchronous.database import Database