remove unintended changes and fix retryable?

Author: sleepyStick

pymongo/asynchronous/bulk.py

@@ -130,24 +130,6 @@ def bulk_ctx_class(self) -> Type[_BulkWriteContext]:
             self.is_encrypted = False
             return _BulkWriteContext
 
-    # @property
-    # def is_retryable(self) -> bool:
-    #     if self.current_run:
-    #         return self.current_run.is_retryable
-    #     return True
-    #
-    # @property
-    # def retrying(self) -> bool:
-    #     if self.current_run:
-    #         return self.current_run.retrying
-    #     return False
-    #
-    # @property
-    # def started_retryable_write(self) -> bool:
-    #     if self.current_run:
-    #         return self.current_run.started_retryable_write
-    #     return False
-
     def add_insert(self, document: _DocumentOut) -> bool:
         """Add an insert document to the list of ops."""
         validate_is_document_type("document", document)
@@ -579,7 +561,6 @@ async def _execute_command(
                     # Start a new retryable write unless one was already
                     # started for this command.
                     if retryable and self.is_retryable and not self.started_retryable_write:
-                        # print("starting retrayable write")
                         session._start_retryable_write()
                         self.started_retryable_write = True
                     session._apply_to(
@@ -596,6 +577,7 @@ async def _execute_command(
                     await self.validate_batch(conn, write_concern)
                 if write_concern.acknowledged:
                     result, to_send = await self._execute_batch(bwc, cmd, ops, client)
+
                     # Retryable writeConcernErrors halt the execution of this run.
                     wce = result.get("writeConcernError", {})
                     if wce.get("code", 0) in _RETRYABLE_ERROR_CODES:
@@ -604,7 +586,9 @@ async def _execute_command(
                         full = copy.deepcopy(full_result)
                         _merge_command(run, full, run.idx_offset, result)
                         _raise_bulk_write_error(full)
+
                     _merge_command(run, full_result, run.idx_offset, result)
+
                     # We're no longer in a retry once a command succeeds.
                     self.retrying = False
                     self.started_retryable_write = False

pymongo/asynchronous/client_bulk.py

@@ -535,7 +535,12 @@ async def _execute_command(
             if session:
                 # Start a new retryable write unless one was already
                 # started for this command.
-                if retryable and self.is_retryable and not self.started_retryable_write:
+                if (
+                    retryable
+                    and self.is_retryable
+                    and not self.started_retryable_write
+                    and not session.in_transaction
+                ):
                     session._start_retryable_write()
                     self.started_retryable_write = True
                 session._apply_to(

pymongo/asynchronous/mongo_client.py

@@ -1947,6 +1947,9 @@ async def _retry_with_session(
         """
         # Ensure that the options supports retry_writes and there is a valid session not in
         # transaction, otherwise, we will not support retry behavior for this txn.
+        retryable = bool(
+            retryable and self.options.retry_writes and session and not session.in_transaction
+        )
         return await self._retry_internal(
             func=func,
             session=session,

pymongo/synchronous/bulk.py

@@ -130,24 +130,6 @@ def bulk_ctx_class(self) -> Type[_BulkWriteContext]:
             self.is_encrypted = False
             return _BulkWriteContext
 
-    # @property
-    # def is_retryable(self) -> bool:
-    #     if self.current_run:
-    #         return self.current_run.is_retryable
-    #     return True
-    #
-    # @property
-    # def retrying(self) -> bool:
-    #     if self.current_run:
-    #         return self.current_run.retrying
-    #     return False
-    #
-    # @property
-    # def started_retryable_write(self) -> bool:
-    #     if self.current_run:
-    #         return self.current_run.started_retryable_write
-    #     return False
-
     def add_insert(self, document: _DocumentOut) -> bool:
         """Add an insert document to the list of ops."""
         validate_is_document_type("document", document)
@@ -579,7 +561,6 @@ def _execute_command(
                     # Start a new retryable write unless one was already
                     # started for this command.
                     if retryable and self.is_retryable and not self.started_retryable_write:
-                        # print("starting retrayable write")
                         session._start_retryable_write()
                         self.started_retryable_write = True
                     session._apply_to(
@@ -596,6 +577,7 @@ def _execute_command(
                     self.validate_batch(conn, write_concern)
                 if write_concern.acknowledged:
                     result, to_send = self._execute_batch(bwc, cmd, ops, client)
+
                     # Retryable writeConcernErrors halt the execution of this run.
                     wce = result.get("writeConcernError", {})
                     if wce.get("code", 0) in _RETRYABLE_ERROR_CODES:
@@ -604,7 +586,9 @@ def _execute_command(
                         full = copy.deepcopy(full_result)
                         _merge_command(run, full, run.idx_offset, result)
                         _raise_bulk_write_error(full)
+
                     _merge_command(run, full_result, run.idx_offset, result)
+
                     # We're no longer in a retry once a command succeeds.
                     self.retrying = False
                     self.started_retryable_write = False

pymongo/synchronous/client_bulk.py

@@ -533,7 +533,12 @@ def _execute_command(
             if session:
                 # Start a new retryable write unless one was already
                 # started for this command.
-                if retryable and self.is_retryable and not self.started_retryable_write:
+                if (
+                    retryable
+                    and self.is_retryable
+                    and not self.started_retryable_write
+                    and not session.in_transaction
+                ):
                     session._start_retryable_write()
                     self.started_retryable_write = True
                 session._apply_to(

pymongo/synchronous/mongo_client.py

@@ -1941,6 +1941,9 @@ def _retry_with_session(
         """
         # Ensure that the options supports retry_writes and there is a valid session not in
         # transaction, otherwise, we will not support retry behavior for this txn.
+        retryable = bool(
+            retryable and self.options.retry_writes and session and not session.in_transaction
+        )
         return self._retry_internal(
             func=func,
             session=session,