o ]#@sdZddlZddlZddlZddlZddlmZddlmZddl m Z m Z m Z ddl mZddlmZddlmZdd lmZmZmZmZmZmZmZdd lmZdd lmZdd lm Z m!Z!dd l"m#Z#Gddde$Z%Gddde$Z&ddZ'Gddde$Z(Gddde$Z)Gddde$Z*ddZ+ddZ,ee-ddgBZ.d Z/d!d"Z0Gd#d$d$e$Z1Gd%d&d&e$Z2Gd'd(d(ej3Z4dS))a Logical sessions for ordering sequential operations. Requires MongoDB 3.6. .. 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. .. mongodoc:: causal-consistency .. _transactions-ref: Transactions ============ 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`. For multi-document transactions, you can only specify read/write (CRUD) operations on existing collections. For example, a multi-document transaction cannot include a create or drop collection/index operations, including an insert operation that would result in the creation of a new collection. A session may only have a single active transaction at a time, multiple transactions on the same session can be executed in sequence. .. versionadded:: 3.7 Sharded Transactions ^^^^^^^^^^^^^^^^^^^^ 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. .. versionadded:: 3.9 .. mongodoc:: transactions Classes ======= N)Binary)Int64)abc integer_typesreraise_instance)SON) Timestamp) monotonic)ConfigurationErrorConnectionFailureInvalidOperationOperationFailure PyMongoErrorServerSelectionTimeoutError WTimeoutError)_RETRYABLE_ERROR_CODES) ReadConcern)ReadPreference _ServerMode) WriteConcernc@s6eZdZdZ  d ddZeddZedd ZdS) SessionOptionsaKOptions for a new :class:`ClientSession`. :Parameters: - `causal_consistency` (optional): If True (the default), read operations are causally ordered within the session. - `default_transaction_options` (optional): The default TransactionOptions to use for transactions started on this session. TNcCs0||_|durt|tstd|f||_dS)Nzedefault_transaction_options must be an instance of pymongo.client_session.TransactionOptions, not: %r)_causal_consistency isinstanceTransactionOptions TypeError_default_transaction_options)selfcausal_consistencydefault_transaction_optionsrH/var/www/html/env/lib/python3.10/site-packages/pymongo/client_session.py__init__s  zSessionOptions.__init__cC|jS)z)Whether causal consistency is configured.)rrrrr rz!SessionOptions.causal_consistencycCr")zThe default TransactionOptions to use for transactions started on this session. .. versionadded:: 3.7 )rr#rrr rsz*SessionOptions.default_transaction_options)TN)__name__ __module__ __qualname____doc__r!propertyrrrrrr rzs   rc@sNeZdZdZ  d ddZeddZeddZed d Zed d Z dS)raOptions for :meth:`ClientSession.start_transaction`. :Parameters: - `read_concern` (optional): 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. - `write_concern` (optional): 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. - `read_preference` (optional): 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`. - `max_commit_time_ms` (optional): 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 NcCs||_||_||_||_|durt|tstd|f|dur6t|ts,td|f|js6t d|f|durFt|t sFtd|f|durSt|t sUtddSdS)NzMread_concern must be an instance of pymongo.read_concern.ReadConcern, not: %rzPwrite_concern must be an instance of pymongo.write_concern.WriteConcern, not: %rzThis transaction's :class:`~pymongo.read_concern.ReadConcern`.)r*r#rrr r/r$zTransactionOptions.read_concerncCr")z@This transaction's :class:`~pymongo.write_concern.WriteConcern`.)r+r#rrr r0r$z TransactionOptions.write_concerncCr")zNThis transaction's :class:`~pymongo.read_preferences.ReadPreference`. )r,r#rrr r1sz"TransactionOptions.read_preferencecCr")zfThe maxTimeMS to use when running a commitTransaction command. .. versionadded:: 3.9 )r-r#rrr r2sz%TransactionOptions.max_commit_time_msNNNN) r%r&r'r(r!r)r/r0r1r2rrrr rs    rcCs.|r|dur|js|jrdStd|f|S)zValidate that an explicit session is not used with an unack'ed write. Returns the session to use for the next operation. NzHExplicit sessions are incompatible with unacknowledged write concern: %r)r. _implicitr )sessionr0rrr _validate_session_write_concernsr6c@(eZdZdZddZddZddZdS) _TransactionContextz;Internal transaction context manager for start_transaction.cCs ||_dSN)_TransactionContext__session)rr5rrr r!s z_TransactionContext.__init__cC|Sr9rr#rrr __enter__ z_TransactionContext.__enter__cCs0|jjr|dur|jdS|jdSdSr9)r:in_transactioncommit_transactionabort_transactionrexc_typeexc_valexc_tbrrr __exit__s z_TransactionContext.__exit__N)r%r&r'r(r!r<rErrrr r8s  r8c@s$eZdZdZdZdZdZdZdZdS) _TxnStateN) r%r&r'NONESTARTING IN_PROGRESS COMMITTEDCOMMITTED_EMPTYABORTEDrrrr rFsrFc@r7) _TransactionzBInternal class to hold transaction information in a ClientSession.cCs$||_tj|_d|_d|_d|_dSNF)optsrFrMstateshardedpinned_addressrecovery_token)rrUrrr r!!s  z_Transaction.__init__cCs|jtjtjfvSr9)rVrFrNrOr#rrr active(z_Transaction.activecCstj|_d|_d|_d|_dSrT)rFrMrVrWrXrYr#rrr reset+s z_Transaction.resetN)r%r&r'r(r!rZr\rrrr rSs  rScCs"|dt|tdddS)zDRe-raise an exception with the UnknownTransactionCommitResult label.UnknownTransactionCommitResultrH)traceN)_add_error_labelrsysexc_infoexcrrr _reraise_with_unknown_commit2s rdcCst|to |jdkS)z/Return true if exc is a MaxTimeMSExpired error.2)rr coderbrrr _max_time_expired_error8srg@rexcCst|tkS)z/Are we within the with_transaction retry limit?)r time"_WITH_TRANSACTION_RETRY_TIME_LIMIT) start_timerrr _within_time_limitKsrmc@s4eZdZdZddZddZddZdd Zd d Zd d Z e ddZ e ddZ e ddZ e ddZe ddZddZ  dAddZ  dAddZdd Zd!d"Zd#d$Zd%d&Zd'd(Zd)d*Zd+d,Zd-d.Zd/d0Ze d1d2Ze d3d4Ze d5d6Zd7d8Zd9d:Z d;d<Z!d=d>Z"d?d@Z#dS)B ClientSessionz-A session for ordering sequential operations.cCs8||_||_||_||_d|_d|_||_td|_dSr9) _client_server_session_options_authset _cluster_time_operation_timer4rS _transaction)rclientserver_sessionoptionsauthsetimplicitrrr r!RszClientSession.__init__cCs|jdddS)zFinish this session. If a transaction has started, abort it. It is an error to use the session after the session has ended. TlockN _end_sessionr#rrr end_session^szClientSession.end_sessionc Csn|jdur5z#|jr|W|j|j|d|_dSW|j|j|d|_dS|j|j|d|_wdSr9)rpr>r@ro_return_server_session)rr|rrr r~es    zClientSession._end_sessioncCs|jdur tddS)NzCannot use ended session)rpr r#rrr _check_endedns zClientSession._check_endedcCr;r9rr#rrr r<rr=zClientSession.__enter__cCs|jdddS)NTr{r}rArrr rEuszClientSession.__exit__cCr")z^The :class:`~pymongo.mongo_client.MongoClient` this session was created from. )ror#rrr rvxzClientSession.clientcCr")z:The :class:`SessionOptions` this session was created with.)rqr#rrr rxr$zClientSession.optionscCs||jjS)z6A BSON document, the opaque server session identifier.)rrp session_idr#rrr rszClientSession.session_idcCr")zZThe cluster time returned by the last operation executed in this session. rsr#rrr cluster_timerzClientSession.cluster_timecCr")z\The operation time returned by the last operation executed in this session. rtr#rrr operation_timerzClientSession.operation_timecCs2|r|S|jj}|ot||}|r|St|j|S)z-Return the inherited TransactionOption value.)rxrgetattrrv)rnamevaltxn_optsrrr _inherit_options zClientSession._inherit_optionNc Cst} |||||z||}Wn(ty;}z|jr"|t|tr6|dr6t |r6WYd}~qd}~ww|jsA|S z| W|Styx}z#|drdt |rdt |sdWYd}~qA|drst |rsWYd}~nd}~wwq)asExecute 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 possiblity 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. 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. :Parameters: - `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. - `read_concern` (optional): The :class:`~pymongo.read_concern.ReadConcern` to use for this transaction. - `write_concern` (optional): The :class:`~pymongo.write_concern.WriteConcern` to use for this transaction. - `read_preference` (optional): 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. :Returns: The return value of the ``callback``. .. versionadded:: 3.9 TTransientTransactionErrorNr]) r rjstart_transaction Exceptionr>r@rrhas_error_labelrmr?rg) rcallbackr/r0r1r2rlretrcrrr with_transactionsRS         zClientSession.with_transactioncCs||jr td|d|}|d|}|d|}|dur*|jj}|r*|j}t|||||j_ |j t j |j_ |t|S)zStart a multi-statement transaction. Takes the same arguments as :class:`TransactionOptions`. .. versionchanged:: 3.9 Added the ``max_commit_time_ms`` option. .. versionadded:: 3.7 zTransaction already in progressr/r0r1N)rr>r rrxrr2rrurUr\rFrNrV_start_retryable_writer8)rr/r0r1r2rUrrr rs&     zClientSession.start_transactionc Csd|d}|jj}|tjurtd|tjtjfvr"tj|j_dS|tjur+td|tj ur7tj |j_d}ztz| d|WnKt y[}z| dt|WYd}~n=d}~wtyp}z t|WYd}~n0d}~wty}z|jtvr}t|WYd}~nd}~wwWtj |j_dSWtj |j_dSWtj |j_dSWtj |j_dStj |j_w)zMCommit a multi-statement transaction. .. versionadded:: 3.7 FNo transaction startedNzz ||dWWYd}~StyT|wd}~ww)aDRun commit or abort with one retry after any retryable error. :Parameters: - `command_name`: Either "commitTransaction" or "abortTransaction". - `explict_retry`: True when this is an explict commit retry attempt, ie the application called session.commit_transaction() twice. TN)_finish_transactionrr r rfr)r command_name explict_retryrcrrr rs*     z,ClientSession._finish_transaction_with_retrycCs|jj}|j}t|dfg}|dkr0|jr|j|d<|r0|j}d|d<|ddtd i|}|jjr:|jj|d<|j |}|j j j ||||d d WdS1sWwYdS) NrGr maxTimeMSmajoritywwtimeouti' recoveryTokenT)r5r0parse_write_concern_errorr) rurUr0rr2document setdefaultrrYro_socket_for_writesadmin_command)rrretryingrUwccmdwc_doc sock_inforrr rs,   $z!ClientSession._finish_transactioncCs@|jdur ||_dS|dur|d|jdkr||_dSdSdS)zInternal cluster time helper.N clusterTimerrrrrr _advance_cluster_times   z#ClientSession._advance_cluster_timecCs:t|tjs tdt|dtstd||dS)zUpdate the cluster time for this session. :Parameters: - `cluster_time`: The :data:`~pymongo.client_session.ClientSession.cluster_time` from another `ClientSession` instance. z6cluster_time must be a subclass of collections.MappingrzInvalid cluster_timeN)rrMappingrgetr ValueErrorrrrrr advance_cluster_times z"ClientSession.advance_cluster_timecCs8|jdur ||_dS|dur||jkr||_dSdSdS)zInternal operation time helper.Nrrrrrr _advance_operation_times    z%ClientSession._advance_operation_timecCs t|ts td||dS)zUpdate the operation time for this session. :Parameters: - `operation_time`: The :data:`~pymongo.client_session.ClientSession.operation_time` from another `ClientSession` instance. z>operation_time must be an instance of bson.timestamp.TimestampN)rrrrrrrr advance_operation_times z$ClientSession.advance_operation_timecCsT||d||d|jr$|jjr&|d}|r(||j_dSdSdSdS)z?Process a response to a command that was run with this session.z $clusterTime operationTimerN)rrrr>rurWrY)rreplyrYrrr _process_responses  zClientSession._process_responsecCs |jduS)z!True if this session is finished.N)rpr#rrr has_ended zClientSession.has_endedcCs |jS)zhTrue if this session has an active multi-statement transaction. .. versionadded:: 3.10 )rurZr#rrr r> zClientSession.in_transactioncCs|jr |jjSdS)z3The mongos address this transaction was created on.N)rurZrXr#rrr _pinned_addresss zClientSession._pinned_addresscCsd|j_|jj|j_dS)z,Pin this session to the given mongos Server.TN)rurW descriptionaddressrX)rserverrrr _pin_mongos szClientSession._pin_mongoscCs d|j_dS)z2Unpin this session from any pinned mongos address.N)rurXr#rrr _unpin_mongoss zClientSession._unpin_mongoscCs|jr|jjjSdS)z3Return read preference of this transaction or None.N)r>rurUr1r#rrr _txn_read_preferences z"ClientSession._txn_read_preferencecCs|t|j_|jj|d<|js|j|r"|jj |d<dS|jro|t j kr1t d|f|jj tjkrctj|j_ d|d<|jjjrM|jjjj}ni}|jjr]|jdur]|j|d<|rc||d<|jj |d<d|d <dSdS) Nlsid txnNumberz9read preference in a transaction must be primary, not: %rTstartTransactionafterClusterTime readConcernF autocommit)rr rjrplast_userr>rur\transaction_idrPRIMARYr rVrFrNrOrUr/rrxrr)rcommand is_retryabler1rcrrr _apply_tos:           zClientSession._apply_tocCs||jdSr9)rrpinc_transaction_idr#rrr r@sz$ClientSession._start_retryable_writer3)$r%r&r'r(r!rr~rr<rEr)rvrxrrrrrrr?r@rrrrrrrrr>rrrrrrrrrr rnPsZ         {  .     'rnc@s8eZdZddZddZddZeddZd d Zd S) _ServerSessioncCs6dttjdi|_t|_d|_d|_ ||_ dS)NidrJrF) ruuiduuid4bytesrr rjr_transaction_iddirtypool_id)rrrrr r!Fs   z_ServerSession.__init__cCs d|_dS)zMark 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. TN)rr#rrr mark_dirtyNrz_ServerSession.mark_dirtycCst|j}||ddkS)NrG<)r rjr)rsession_timeout_minutes idle_secondsrrr timed_outVsz_ServerSession.timed_outcCs t|jS)zPositive 64-bit integer.)rrr#rrr r\rz_ServerSession.transaction_idcCs|jd7_dSNrG)rr#rrr rar[z!_ServerSession.inc_transaction_idN) r%r&r'r!rrr)rrrrrr rEs  rcsPeZdZdZfddZddZddZdd Zd d Zd d Z ddZ Z S)_ServerSessionPoolzsPool of _ServerSession objects. This class is not thread-safe, access it while holding the Topology lock. cs tt|j|i|d|_dS)Nr)superrr!r)rargskwargs __class__rr r!js z_ServerSessionPool.__init__cCs|jd7_|dSr)rclearr#rrr r\ns z_ServerSessionPool.resetcCs g}|r||j|s|Sr9)appendpopr)ridsrrr pop_allrs z_ServerSessionPool.pop_allcCs2|||r|}||s|S|st|jSr9) _clear_stalepopleftrrr)rrsrrr get_server_sessionxs   z%_ServerSessionPool.get_server_sessioncCs&||||s||dSdSr9)rrreturn_server_session_no_lock)rrwrrrr return_server_sessions  z(_ServerSessionPool.return_server_sessioncCs(|j|jkr|js||dSdSdSr9)rr appendleft)rrwrrr rsz0_ServerSessionPool.return_server_session_no_lockcCs,|r|d|r|ndS|sdSdS)N)rr)rrrrr rs   z_ServerSessionPool._clear_stale) r%r&r'r(r!r\rrrrr __classcell__rrrr res r)5r( collectionsosr`r bson.binaryr bson.int64rbson.py3compatrrrbson.sonrbson.timestamprpymongor pymongo.errorsr r r r rrrpymongo.helpersrpymongo.read_concernrpymongo.read_preferencesrrpymongo.write_concernrobjectrrr6r8rFrSrdrg frozensetrrkrmrnrdequerrrrr sHR     $   $R  x