# Copyright 2017 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. """Logical sessions for ordering sequential operations. .. versionadded:: 3.6 Causally Consistent Reads ========================= .. code-block:: python with client.start_session(causal_consistency=True) as session: collection = client.db.collection collection.update_one({"_id": 1}, {"$set": {"x": 10}}, session=session) secondary_c = collection.with_options(read_preference=ReadPreference.SECONDARY) # A secondary read waits for replication of the write. secondary_c.find_one({"_id": 1}, session=session) If `causal_consistency` is True (the default), read operations that use the session are causally after previous read and write operations. Using a causally consistent session, an application can read its own writes and is guaranteed monotonic reads, even when reading from replica set secondaries. .. seealso:: The MongoDB documentation on `causal-consistency `_. .. _transactions-ref: Transactions ============ .. versionadded:: 3.7 MongoDB 4.0 adds support for transactions on replica set primaries. A transaction is associated with a :class:`ClientSession`. To start a transaction on a session, use :meth:`ClientSession.start_transaction` in a with-statement. Then, execute an operation within the transaction by passing the session to the operation: .. code-block:: python orders = client.db.orders inventory = client.db.inventory with client.start_session() as session: with session.start_transaction(): orders.insert_one({"sku": "abc123", "qty": 100}, session=session) inventory.update_one( {"sku": "abc123", "qty": {"$gte": 100}}, {"$inc": {"qty": -100}}, session=session, ) Upon normal completion of ``with session.start_transaction()`` block, the transaction automatically calls :meth:`ClientSession.commit_transaction`. If the block exits with an exception, the transaction automatically calls :meth:`ClientSession.abort_transaction`. In general, multi-document transactions only support read/write (CRUD) operations on existing collections. However, MongoDB 4.4 adds support for creating collections and indexes with some limitations, including an insert operation that would result in the creation of a new collection. For a complete description of all the supported and unsupported operations see the `MongoDB server's documentation for transactions `_. A session may only have a single active transaction at a time, multiple transactions on the same session can be executed in sequence. Sharded Transactions ^^^^^^^^^^^^^^^^^^^^ .. versionadded:: 3.9 PyMongo 3.9 adds support for transactions on sharded clusters running MongoDB >=4.2. Sharded transactions have the same API as replica set transactions. When running a transaction against a sharded cluster, the session is pinned to the mongos server selected for the first operation in the transaction. All subsequent operations that are part of the same transaction are routed to the same mongos server. When the transaction is completed, by running either commitTransaction or abortTransaction, the session is unpinned. .. seealso:: The MongoDB documentation on `transactions `_. .. _snapshot-reads-ref: Snapshot Reads ============== .. versionadded:: 3.12 MongoDB 5.0 adds support for snapshot reads. Snapshot reads are requested by passing the ``snapshot`` option to :meth:`~pymongo.mongo_client.MongoClient.start_session`. If ``snapshot`` is True, all read operations that use this session read data from the same snapshot timestamp. The server chooses the latest majority-committed snapshot timestamp when executing the first read operation using the session. Subsequent reads on this session read from the same snapshot timestamp. Snapshot reads are also supported when reading from replica set secondaries. .. code-block:: python # Each read using this session reads data from the same point in time. with client.start_session(snapshot=True) as session: order = orders.find_one({"sku": "abc123"}, session=session) inventory = inventory.find_one({"sku": "abc123"}, session=session) Snapshot Reads Limitations ^^^^^^^^^^^^^^^^^^^^^^^^^^ Snapshot reads sessions are incompatible with ``causal_consistency=True``. Only the following read operations are supported in a snapshot reads session: - :meth:`~pymongo.collection.Collection.find` - :meth:`~pymongo.collection.Collection.find_one` - :meth:`~pymongo.collection.Collection.aggregate` - :meth:`~pymongo.collection.Collection.count_documents` - :meth:`~pymongo.collection.Collection.distinct` (on unsharded collections) Classes ======= """ from __future__ import annotations import collections import time import uuid from collections.abc import Mapping as _Mapping from typing import ( TYPE_CHECKING, Any, Callable, ContextManager, Mapping, MutableMapping, NoReturn, Optional, Type, TypeVar, ) from bson.binary import Binary 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, InvalidOperation, OperationFailure, PyMongoError, WTimeoutError, ) from pymongo.helpers 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.write_concern import WriteConcern if TYPE_CHECKING: from types import TracebackType from pymongo.pool import Connection from pymongo.server import Server from pymongo.typings import ClusterTime, _Address 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 " "pymongo.client_session.TransactionOptions, not: {!r}".format( default_transaction_options ) ) 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( "transactions do not support unacknowledged write concern" f": {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("max_commit_time_ms must be an integer or None") @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.""" def __init__(self, session: ClientSession): self.__session = session def __enter__(self) -> _TransactionContext: return self def __exit__( self, exc_type: Optional[Type[BaseException]], exc_val: Optional[BaseException], exc_tb: Optional[TracebackType], ) -> None: if self.__session.in_transaction: if exc_val is None: self.__session.commit_transaction() else: 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.""" def __init__(self, opts: Optional[TransactionOptions], client: MongoClient): self.opts = opts self.state = _TxnState.NONE self.sharded = False self.pinned_address: Optional[_Address] = None self.conn_mgr: Optional[_ConnectionManager] = None self.recovery_token = None self.attempt = 0 self.client = client def active(self) -> bool: return self.state in (_TxnState.STARTING, _TxnState.IN_PROGRESS) def starting(self) -> bool: return self.state == _TxnState.STARTING @property def pinned_conn(self) -> Optional[Connection]: if self.active() and self.conn_mgr: return self.conn_mgr.conn return None def pin(self, server: Server, conn: Connection) -> None: self.sharded = True self.pinned_address = server.description.address if server.description.server_type == SERVER_TYPE.LoadBalancer: conn.pin_txn() self.conn_mgr = _ConnectionManager(conn, False) def unpin(self) -> None: self.pinned_address = None if self.conn_mgr: self.conn_mgr.close() self.conn_mgr = None def reset(self) -> None: self.unpin() self.state = _TxnState.NONE self.sharded = False self.recovery_token = None self.attempt = 0 def __del__(self) -> None: if self.conn_mgr: # Reuse the cursor closing machinery to return the socket to the # pool soon. self.client._close_cursor_soon(0, None, self.conn_mgr) 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 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 # WriteConcernFailed. _UNKNOWN_COMMIT_ERROR_CODES: frozenset = _RETRYABLE_ERROR_CODES | frozenset( [ 64, # WriteConcernFailed 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 def _within_time_limit(start_time: float) -> bool: """Are we within the with_transaction retry limit?""" return time.monotonic() - start_time < _WITH_TRANSACTION_RETRY_TIME_LIMIT _T = TypeVar("_T") if TYPE_CHECKING: from pymongo.mongo_client import MongoClient class ClientSession: """A session for ordering sequential operations. :class:`ClientSession` instances are **not thread-safe or fork-safe**. They can only be used by one thread or process at a time. A single :class:`ClientSession` cannot be used to run multiple operations concurrently. Should not be initialized directly by application developers - to create a :class:`ClientSession`, call :meth:`~pymongo.mongo_client.MongoClient.start_session`. """ def __init__( self, client: MongoClient, server_session: Any, options: SessionOptions, implicit: bool, ) -> None: # A MongoClient, a _ServerSession, a SessionOptions, and a set. self._client: MongoClient = client self._server_session = server_session self._options = options self._cluster_time: Optional[Mapping[str, Any]] = None self._operation_time: Optional[Timestamp] = None self._snapshot_time = None # Is this an implicitly created session? self._implicit = implicit self._transaction = _Transaction(None, client) def end_session(self) -> None: """Finish this session. If a transaction has started, abort it. It is an error to use the session after the session has ended. """ if self._server_session is not None: try: if self.in_transaction: self.abort_transaction() # It's possible we're still pinned here when the transaction # is in the committed state when the session is discarded. self._unpin() finally: 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") def __enter__(self) -> ClientSession: return self def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None: self.end_session() @property def client(self) -> MongoClient: """The :class:`~pymongo.mongo_client.MongoClient` this session was created from. """ return self._client @property def options(self) -> SessionOptions: """The :class:`SessionOptions` this session was created with.""" return self._options @property def session_id(self) -> Mapping[str, Any]: """A BSON document, the opaque server session identifier.""" self._check_ended() self._materialize(self._client.topology_description.logical_session_timeout_minutes) return self._server_session.session_id @property def _transaction_id(self) -> Int64: """The current transaction id for the underlying server session.""" self._materialize(self._client.topology_description.logical_session_timeout_minutes) return self._server_session.transaction_id @property def cluster_time(self) -> Optional[ClusterTime]: """The cluster time returned by the last operation executed in this session. """ return self._cluster_time @property def operation_time(self) -> Optional[Timestamp]: """The operation time returned by the last operation executed in this session. """ return self._operation_time def _inherit_option(self, name: str, val: _T) -> _T: """Return the inherited TransactionOption value.""" if val: return val txn_opts = self.options.default_transaction_options parent_val = txn_opts and getattr(txn_opts, name) if parent_val: return parent_val return getattr(self.client, name) def with_transaction( self, callback: Callable[[ClientSession], _T], read_concern: Optional[ReadConcern] = None, write_concern: Optional[WriteConcern] = None, read_preference: Optional[_ServerMode] = None, max_commit_time_ms: Optional[int] = None, ) -> _T: """Execute a callback in a transaction. This method starts a transaction on this session, executes ``callback`` once, and then commits the transaction. For example:: def callback(session): orders = session.client.db.orders inventory = session.client.db.inventory orders.insert_one({"sku": "abc123", "qty": 100}, session=session) inventory.update_one({"sku": "abc123", "qty": {"$gte": 100}}, {"$inc": {"qty": -100}}, session=session) with client.start_session() as session: session.with_transaction(callback) To pass arbitrary arguments to the ``callback``, wrap your callable with a ``lambda`` like this:: def callback(session, custom_arg, custom_kwarg=None): # Transaction operations... with client.start_session() as session: session.with_transaction( lambda s: callback(s, "custom_arg", custom_kwarg=1)) In the event of an exception, ``with_transaction`` may retry the commit or the entire transaction, therefore ``callback`` may be invoked multiple times by a single call to ``with_transaction``. Developers should be mindful of this possibility when writing a ``callback`` that modifies application state or has any other side-effects. Note that even when the ``callback`` is invoked multiple times, ``with_transaction`` ensures that the transaction will be committed at-most-once on the server. The ``callback`` should not attempt to start new transactions, but should simply run operations meant to be contained within a transaction. The ``callback`` should also not commit the transaction; this is handled automatically by ``with_transaction``. If the ``callback`` does commit or abort the transaction without error, however, ``with_transaction`` will return without taking further action. :class:`ClientSession` instances are **not thread-safe or fork-safe**. Consequently, the ``callback`` must not attempt to execute multiple operations concurrently. When ``callback`` raises an exception, ``with_transaction`` automatically aborts the current transaction. When ``callback`` or :meth:`~ClientSession.commit_transaction` raises an exception that includes the ``"TransientTransactionError"`` error label, ``with_transaction`` starts a new transaction and re-executes the ``callback``. When :meth:`~ClientSession.commit_transaction` raises an exception with the ``"UnknownTransactionCommitResult"`` error label, ``with_transaction`` retries the commit until the result of the transaction is known. This method will cease retrying after 120 seconds has elapsed. This timeout is not configurable and any exception raised by the ``callback`` or by :meth:`ClientSession.commit_transaction` after the timeout is reached will be re-raised. Applications that desire a different timeout duration should not use this method. :param callback: The callable ``callback`` to run inside a transaction. The callable must accept a single argument, this session. Note, under certain error conditions the callback may be run multiple times. :param read_concern: The :class:`~pymongo.read_concern.ReadConcern` to use for this transaction. :param write_concern: The :class:`~pymongo.write_concern.WriteConcern` to use for this transaction. :param read_preference: The read preference to use for this transaction. If ``None`` (the default) the :attr:`read_preference` of this :class:`Database` is used. See :mod:`~pymongo.read_preferences` for options. :return: The return value of the ``callback``. .. versionadded:: 3.9 """ start_time = time.monotonic() while True: self.start_transaction(read_concern, write_concern, read_preference, max_commit_time_ms) try: ret = callback(self) except Exception as exc: if self.in_transaction: self.abort_transaction() if ( isinstance(exc, PyMongoError) and exc.has_error_label("TransientTransactionError") and _within_time_limit(start_time) ): # Retry the entire transaction. continue raise if not self.in_transaction: # Assume callback intentionally ended the transaction. return ret while True: try: self.commit_transaction() except PyMongoError as exc: if ( exc.has_error_label("UnknownTransactionCommitResult") and _within_time_limit(start_time) and not _max_time_expired_error(exc) ): # Retry the commit. continue if exc.has_error_label("TransientTransactionError") and _within_time_limit( start_time ): # Retry the entire transaction. break raise # Commit succeeded. return ret def start_transaction( self, read_concern: Optional[ReadConcern] = None, write_concern: Optional[WriteConcern] = None, read_preference: Optional[_ServerMode] = None, max_commit_time_ms: Optional[int] = None, ) -> ContextManager: """Start a multi-statement transaction. Takes the same arguments as :class:`TransactionOptions`. .. versionchanged:: 3.9 Added the ``max_commit_time_ms`` option. .. versionadded:: 3.7 """ self._check_ended() if self.options.snapshot: raise InvalidOperation("Transactions are not supported in snapshot sessions") if self.in_transaction: raise InvalidOperation("Transaction already in progress") read_concern = self._inherit_option("read_concern", read_concern) write_concern = self._inherit_option("write_concern", write_concern) read_preference = self._inherit_option("read_preference", read_preference) if max_commit_time_ms is None: opts = self.options.default_transaction_options if opts: max_commit_time_ms = opts.max_commit_time_ms self._transaction.opts = TransactionOptions( read_concern, write_concern, read_preference, max_commit_time_ms ) self._transaction.reset() self._transaction.state = _TxnState.STARTING self._start_retryable_write() return _TransactionContext(self) def commit_transaction(self) -> None: """Commit a multi-statement transaction. .. versionadded:: 3.7 """ self._check_ended() state = self._transaction.state if state is _TxnState.NONE: raise InvalidOperation("No transaction started") elif state in (_TxnState.STARTING, _TxnState.COMMITTED_EMPTY): # Server transaction was never started, no need to send a command. self._transaction.state = _TxnState.COMMITTED_EMPTY return elif state is _TxnState.ABORTED: raise InvalidOperation("Cannot call commitTransaction after calling abortTransaction") elif state is _TxnState.COMMITTED: # We're explicitly retrying the commit, move the state back to # "in progress" so that in_transaction returns true. self._transaction.state = _TxnState.IN_PROGRESS try: self._finish_transaction_with_retry("commitTransaction") except ConnectionFailure as exc: # We do not know if the commit was successfully applied on the # server or if it satisfied the provided write concern, set the # unknown commit error label. exc._remove_error_label("TransientTransactionError") _reraise_with_unknown_commit(exc) except WTimeoutError as exc: # We do not know if the commit has satisfied the provided write # concern, add the unknown commit error label. _reraise_with_unknown_commit(exc) except OperationFailure as exc: if exc.code not in _UNKNOWN_COMMIT_ERROR_CODES: # The server reports errorLabels in the case. raise # We do not know if the commit was successfully applied on the # server or if it satisfied the provided write concern, set the # unknown commit error label. _reraise_with_unknown_commit(exc) finally: self._transaction.state = _TxnState.COMMITTED def abort_transaction(self) -> None: """Abort a multi-statement transaction. .. versionadded:: 3.7 """ self._check_ended() state = self._transaction.state if state is _TxnState.NONE: raise InvalidOperation("No transaction started") elif state is _TxnState.STARTING: # Server transaction was never started, no need to send a command. self._transaction.state = _TxnState.ABORTED return elif state is _TxnState.ABORTED: raise InvalidOperation("Cannot call abortTransaction twice") elif state in (_TxnState.COMMITTED, _TxnState.COMMITTED_EMPTY): raise InvalidOperation("Cannot call abortTransaction after calling commitTransaction") try: self._finish_transaction_with_retry("abortTransaction") except (OperationFailure, ConnectionFailure): # The transactions spec says to ignore abortTransaction errors. pass finally: self._transaction.state = _TxnState.ABORTED self._unpin() def _finish_transaction_with_retry(self, command_name: str) -> dict[str, Any]: """Run commit or abort with one retry after any retryable error. :param command_name: Either "commitTransaction" or "abortTransaction". """ def func( _session: Optional[ClientSession], conn: Connection, _retryable: bool ) -> dict[str, Any]: return self._finish_transaction(conn, command_name) return self._client._retry_internal(func, self, None, retryable=True, operation=_Op.ABORT) def _finish_transaction(self, conn: Connection, command_name: str) -> dict[str, Any]: self._transaction.attempt += 1 opts = self._transaction.opts assert opts wc = opts.write_concern cmd = {command_name: 1} if command_name == "commitTransaction": if opts.max_commit_time_ms and _csot.get_timeout() is None: cmd["maxTimeMS"] = opts.max_commit_time_ms # Transaction spec says that after the initial commit attempt, # subsequent commitTransaction commands should be upgraded to use # w:"majority" and set a default value of 10 seconds for wtimeout. if self._transaction.attempt > 1: assert wc wc_doc = wc.document wc_doc["w"] = "majority" wc_doc.setdefault("wtimeout", 10000) wc = WriteConcern(**wc_doc) if self._transaction.recovery_token: cmd["recoveryToken"] = self._transaction.recovery_token return self._client.admin._command( conn, cmd, session=self, write_concern=wc, parse_write_concern_error=True ) def _advance_cluster_time(self, cluster_time: Optional[Mapping[str, Any]]) -> None: """Internal cluster time helper.""" if self._cluster_time is None: self._cluster_time = cluster_time elif cluster_time is not None: if cluster_time["clusterTime"] > self._cluster_time["clusterTime"]: self._cluster_time = cluster_time def advance_cluster_time(self, cluster_time: Mapping[str, Any]) -> None: """Update the cluster time for this session. :param cluster_time: The :data:`~pymongo.client_session.ClientSession.cluster_time` from another `ClientSession` instance. """ if not isinstance(cluster_time, _Mapping): raise TypeError("cluster_time must be a subclass of collections.Mapping") if not isinstance(cluster_time.get("clusterTime"), Timestamp): raise ValueError("Invalid cluster_time") self._advance_cluster_time(cluster_time) def _advance_operation_time(self, operation_time: Optional[Timestamp]) -> None: """Internal operation time helper.""" if self._operation_time is None: self._operation_time = operation_time elif operation_time is not None: if operation_time > self._operation_time: self._operation_time = operation_time def advance_operation_time(self, operation_time: Timestamp) -> None: """Update the operation time for this session. :param operation_time: The :data:`~pymongo.client_session.ClientSession.operation_time` from another `ClientSession` instance. """ if not isinstance(operation_time, Timestamp): raise TypeError("operation_time must be an instance of bson.timestamp.Timestamp") self._advance_operation_time(operation_time) def _process_response(self, reply: Mapping[str, Any]) -> None: """Process a response to a command that was run with this session.""" self._advance_cluster_time(reply.get("$clusterTime")) self._advance_operation_time(reply.get("operationTime")) if self._options.snapshot and self._snapshot_time is None: if "cursor" in reply: ct = reply["cursor"].get("atClusterTime") else: ct = reply.get("atClusterTime") self._snapshot_time = ct if self.in_transaction and self._transaction.sharded: recovery_token = reply.get("recoveryToken") if recovery_token: self._transaction.recovery_token = recovery_token @property def has_ended(self) -> bool: """True if this session is finished.""" return self._server_session is None @property def in_transaction(self) -> bool: """True if this session has an active multi-statement transaction. .. versionadded:: 3.10 """ return self._transaction.active() @property def _starting_transaction(self) -> bool: """True if this session is starting a multi-statement transaction.""" return self._transaction.starting() @property def _pinned_address(self) -> Optional[_Address]: """The mongos address this transaction was created on.""" if self._transaction.active(): return self._transaction.pinned_address return None @property def _pinned_connection(self) -> Optional[Connection]: """The connection this transaction was started on.""" return self._transaction.pinned_conn def _pin(self, server: Server, conn: Connection) -> None: """Pin this session to the given Server or to the given connection.""" self._transaction.pin(server, conn) def _unpin(self) -> None: """Unpin this session from any pinned Server.""" self._transaction.unpin() def _txn_read_preference(self) -> Optional[_ServerMode]: """Return read preference of this transaction or None.""" if self.in_transaction: assert self._transaction.opts return self._transaction.opts.read_preference return None def _materialize(self, logical_session_timeout_minutes: Optional[int] = None) -> None: if isinstance(self._server_session, _EmptyServerSession): old = self._server_session self._server_session = self._client._topology.get_server_session( logical_session_timeout_minutes ) if old.started_retryable_write: self._server_session.inc_transaction_id() def _apply_to( self, command: MutableMapping[str, Any], is_retryable: bool, read_preference: _ServerMode, conn: Connection, ) -> None: if not conn.supports_sessions: if not self._implicit: raise ConfigurationError("Sessions are not supported by this MongoDB deployment") return self._check_ended() self._materialize(conn.logical_session_timeout_minutes) if self.options.snapshot: self._update_read_concern(command, conn) self._server_session.last_use = time.monotonic() command["lsid"] = self._server_session.session_id if is_retryable: command["txnNumber"] = self._server_session.transaction_id return if self.in_transaction: if read_preference != ReadPreference.PRIMARY: raise InvalidOperation( f"read preference in a transaction must be primary, not: {read_preference!r}" ) if self._transaction.state == _TxnState.STARTING: # First command begins a new transaction. self._transaction.state = _TxnState.IN_PROGRESS command["startTransaction"] = True assert self._transaction.opts if self._transaction.opts.read_concern: rc = self._transaction.opts.read_concern.document if rc: command["readConcern"] = rc self._update_read_concern(command, conn) command["txnNumber"] = self._server_session.transaction_id command["autocommit"] = False def _start_retryable_write(self) -> None: self._check_ended() self._server_session.inc_transaction_id() def _update_read_concern(self, cmd: MutableMapping[str, Any], conn: Connection) -> None: if self.options.causal_consistency and self.operation_time is not None: cmd.setdefault("readConcern", {})["afterClusterTime"] = self.operation_time if self.options.snapshot: if conn.max_wire_version < 13: raise ConfigurationError("Snapshot reads require MongoDB 5.0 or later") rc = cmd.setdefault("readConcern", {}) rc["level"] = "snapshot" if self._snapshot_time is not None: rc["atClusterTime"] = self._snapshot_time 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): """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