bring back retryable var cuz default of false isn't great and move retryable var to bulk class to match client_bulk

Author: sleepyStick

pymongo/asynchronous/bulk.py

@@ -258,7 +258,7 @@ def gen_ordered(
                 yield run
                 run = _Run(op_type)
             run.add(idx, operation)
-            self.is_retryable = self.is_retryable and retryable
+            run.is_retryable = run.is_retryable and retryable
         if run is None:
             raise InvalidOperation("No operations to execute")
         yield run
@@ -276,7 +276,7 @@ def gen_unordered(
             retryable = process(request)
             (op_type, operation) = self.ops[idx]
             operations[op_type].add(idx, operation)
-            self.is_retryable = self.is_retryable and retryable
+            operations[op_type].is_retryable = operations[op_type].is_retryable and retryable
         if (
             len(operations[_INSERT].ops) == 0
             and len(operations[_UPDATE].ops) == 0
@@ -517,6 +517,7 @@ async def _execute_command(
         session: Optional[AsyncClientSession],
         conn: AsyncConnection,
         op_id: int,
+        retryable: bool,
         full_result: MutableMapping[str, Any],
         validate: bool,
         final_write_concern: Optional[WriteConcern] = None,
@@ -536,6 +537,9 @@ async def _execute_command(
         last_run = False
 
         while run:
+            self.is_retryable = run.is_retryable
+            self.retrying = run.retrying
+            self.started_retryable_write = run.started_retryable_write
             if not self.retrying:
                 self.next_run = next(generator, None)
                 if self.next_run is None:
@@ -570,10 +574,13 @@ async def _execute_command(
                 if session:
                     # Start a new retryable write unless one was already
                     # started for this command.
-                    if self.is_retryable and not self.started_retryable_write:
+                    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(cmd, self.is_retryable, ReadPreference.PRIMARY, conn)
+                    session._apply_to(
+                        cmd, retryable and self.is_retryable, ReadPreference.PRIMARY, conn
+                    )
                 conn.send_cluster_time(cmd, session, client)
                 conn.add_server_api(cmd)
                 # CSOT: apply timeout before encoding the command.
@@ -593,12 +600,10 @@ 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.
-                    run.retrying = False
-                    run.started_retryable_write = False
+                    self.retrying = False
+                    self.started_retryable_write = False
 
                     if self.ordered and "writeErrors" in result:
                         break
@@ -636,27 +641,29 @@ async def execute_command(
         op_id = _randint()
 
         async def retryable_bulk(
-            session: Optional[AsyncClientSession],
-            conn: AsyncConnection,
+            session: Optional[AsyncClientSession], conn: AsyncConnection, retryable: bool
         ) -> None:
             await self._execute_command(
                 generator,
                 write_concern,
                 session,
                 conn,
                 op_id,
+                retryable,
                 full_result,
                 validate=False,
             )
 
         client = self.collection.database.client
         _ = await client._retryable_write(
+            self.is_retryable,
             retryable_bulk,
             session,
             operation,
             bulk=self,  # type: ignore[arg-type]
             operation_id=op_id,
         )
+
         if full_result["writeErrors"] or full_result["writeConcernErrors"]:
             _raise_bulk_write_error(full_result)
         return full_result
@@ -730,6 +737,7 @@ async def execute_command_no_results(
                 None,
                 conn,
                 op_id,
+                False,
                 full_result,
                 True,
                 write_concern,

pymongo/asynchronous/client_bulk.py

@@ -489,6 +489,7 @@ async def _execute_command(
         session: Optional[AsyncClientSession],
         conn: AsyncConnection,
         op_id: int,
+        retryable: bool,
         full_result: MutableMapping[str, Any],
         final_write_concern: Optional[WriteConcern] = None,
     ) -> None:
@@ -534,10 +535,12 @@ async def _execute_command(
             if session:
                 # Start a new retryable write unless one was already
                 # started for this command.
-                if self.is_retryable and not self.started_retryable_write:
+                if retryable and self.is_retryable and not self.started_retryable_write:
                     session._start_retryable_write()
                     self.started_retryable_write = True
-                session._apply_to(cmd, self.is_retryable, ReadPreference.PRIMARY, conn)
+                session._apply_to(
+                    cmd, retryable and self.is_retryable, ReadPreference.PRIMARY, conn
+                )
             conn.send_cluster_time(cmd, session, self.client)
             conn.add_server_api(cmd)
             # CSOT: apply timeout before encoding the command.
@@ -564,7 +567,11 @@ async def _execute_command(
 
                     # Synthesize the full bulk result without modifying the
                     # current one because this write operation may be retried.
-                    if self.is_retryable and (retryable_top_level_error or retryable_network_error):
+                    if (
+                        retryable
+                        and self.is_retryable
+                        and (retryable_top_level_error or retryable_network_error)
+                    ):
                         full = copy.deepcopy(full_result)
                         _merge_command(self.ops, self.idx_offset, full, result)
                         _throw_client_bulk_write_exception(full, self.verbose_results)
@@ -583,7 +590,7 @@ async def _execute_command(
                     _merge_command(self.ops, self.idx_offset, full_result, result)
                     break
 
-                if self.is_retryable:
+                if retryable and self.is_retryable:
                     # Retryable writeConcernErrors halt the execution of this batch.
                     wce = result.get("writeConcernError", {})
                     if wce.get("code", 0) in _RETRYABLE_ERROR_CODES:
@@ -638,6 +645,7 @@ async def execute_command(
         async def retryable_bulk(
             session: Optional[AsyncClientSession],
             conn: AsyncConnection,
+            retryable: bool,
         ) -> None:
             if conn.max_wire_version < 25:
                 raise InvalidOperation(
@@ -648,10 +656,12 @@ async def retryable_bulk(
                 session,
                 conn,
                 op_id,
+                retryable,
                 full_result,
             )
 
         await self.client._retryable_write(
+            self.is_retryable,
             retryable_bulk,
             session,
             operation,

pymongo/asynchronous/client_session.py

@@ -854,12 +854,13 @@ async def _finish_transaction_with_retry(self, command_name: str) -> dict[str, A
         """
 
         async def func(
-            _session: Optional[AsyncClientSession],
-            conn: AsyncConnection,
+            _session: Optional[AsyncClientSession], conn: AsyncConnection, _retryable: bool
         ) -> dict[str, Any]:
             return await self._finish_transaction(conn, command_name)
 
-        return await self._client._retry_internal(func, self, None, operation=_Op.ABORT)
+        return await self._client._retry_internal(
+            func, self, None, retryable=True, operation=_Op.ABORT
+        )
 
     async def _finish_transaction(self, conn: AsyncConnection, command_name: str) -> dict[str, Any]:
         self._transaction.attempt += 1

pymongo/asynchronous/collection.py

@@ -785,7 +785,7 @@ async def bulk_write(
 
         write_concern = self._write_concern_for(session)
 
-        def process_for_bulk(request: _WriteOp) -> bool:
+        def process_for_bulk(request: Union[_DocumentType, RawBSONDocument, _WriteOp]) -> bool:
             try:
                 return request._add_to_bulk(blk)
             except AttributeError:
@@ -810,27 +810,32 @@ async def _insert_one(
     ) -> Any:
         """Internal helper for inserting a single document."""
         write_concern = write_concern or self.write_concern
+        acknowledged = write_concern.acknowledged
         command = {"insert": self.name, "ordered": ordered, "documents": [doc]}
         if comment is not None:
             command["comment"] = comment
 
         async def _insert_command(
-            session: Optional[AsyncClientSession], conn: AsyncConnection
+            session: Optional[AsyncClientSession], conn: AsyncConnection, retryable_write: bool
         ) -> None:
             if bypass_doc_val is not None:
                 command["bypassDocumentValidation"] = bypass_doc_val
+
             result = await conn.command(
                 self._database.name,
                 command,
                 write_concern=write_concern,
                 codec_options=self._write_response_codec_options,
                 session=session,
                 client=self._database.client,
+                retryable_write=retryable_write,
             )
 
             _check_write_command_response(result)
 
-        await self._database.client._retryable_write(_insert_command, session, operation=_Op.INSERT)
+        await self._database.client._retryable_write(
+            acknowledged, _insert_command, session, operation=_Op.INSERT
+        )
 
         if not isinstance(doc, RawBSONDocument):
             return doc.get("_id")
@@ -959,7 +964,7 @@ async def insert_many(
             raise TypeError("documents must be a non-empty list")
         inserted_ids: list[ObjectId] = []
 
-        def process_for_bulk(document: Union[_DocumentType, RawBSONDocument]) -> bool:
+        def process_for_bulk(document: Union[_DocumentType, RawBSONDocument, _WriteOp]) -> bool:
             """A generator that validates documents and handles _ids."""
             common.validate_is_document_type("document", document)
             if not isinstance(document, RawBSONDocument):
@@ -989,6 +994,7 @@ async def _update(
         array_filters: Optional[Sequence[Mapping[str, Any]]] = None,
         hint: Optional[_IndexKeyHint] = None,
         session: Optional[AsyncClientSession] = None,
+        retryable_write: bool = False,
         let: Optional[Mapping[str, Any]] = None,
         sort: Optional[Mapping[str, Any]] = None,
         comment: Optional[Any] = None,
@@ -1051,6 +1057,7 @@ async def _update(
                 codec_options=self._write_response_codec_options,
                 session=session,
                 client=self._database.client,
+                retryable_write=retryable_write,
             )
         ).copy()
         _check_write_command_response(result)
@@ -1090,7 +1097,7 @@ async def _update_retryable(
         """Internal update / replace helper."""
 
         async def _update(
-            session: Optional[AsyncClientSession], conn: AsyncConnection
+            session: Optional[AsyncClientSession], conn: AsyncConnection, retryable_write: bool
         ) -> Optional[Mapping[str, Any]]:
             return await self._update(
                 conn,
@@ -1106,12 +1113,14 @@ async def _update(
                 array_filters=array_filters,
                 hint=hint,
                 session=session,
+                retryable_write=retryable_write,
                 let=let,
                 sort=sort,
                 comment=comment,
             )
 
         return await self._database.client._retryable_write(
+            (write_concern or self.write_concern).acknowledged and not multi,
             _update,
             session,
             operation,
@@ -1501,6 +1510,7 @@ async def _delete(
         collation: Optional[_CollationIn] = None,
         hint: Optional[_IndexKeyHint] = None,
         session: Optional[AsyncClientSession] = None,
+        retryable_write: bool = False,
         let: Optional[Mapping[str, Any]] = None,
         comment: Optional[Any] = None,
     ) -> Mapping[str, Any]:
@@ -1540,6 +1550,7 @@ async def _delete(
             codec_options=self._write_response_codec_options,
             session=session,
             client=self._database.client,
+            retryable_write=retryable_write,
         )
         _check_write_command_response(result)
         return result
@@ -1560,7 +1571,7 @@ async def _delete_retryable(
         """Internal delete helper."""
 
         async def _delete(
-            session: Optional[AsyncClientSession], conn: AsyncConnection
+            session: Optional[AsyncClientSession], conn: AsyncConnection, retryable_write: bool
         ) -> Mapping[str, Any]:
             return await self._delete(
                 conn,
@@ -1572,11 +1583,13 @@ async def _delete(
                 collation=collation,
                 hint=hint,
                 session=session,
+                retryable_write=retryable_write,
                 let=let,
                 comment=comment,
             )
 
         return await self._database.client._retryable_write(
+            (write_concern or self.write_concern).acknowledged and not multi,
             _delete,
             session,
             operation=_Op.DELETE,
@@ -3221,7 +3234,7 @@ async def _find_and_modify(
         write_concern = self._write_concern_for_cmd(cmd, session)
 
         async def _find_and_modify_helper(
-            session: Optional[AsyncClientSession], conn: AsyncConnection
+            session: Optional[AsyncClientSession], conn: AsyncConnection, retryable_write: bool
         ) -> Any:
             acknowledged = write_concern.acknowledged
             if array_filters is not None:
@@ -3247,13 +3260,15 @@ async def _find_and_modify_helper(
                 write_concern=write_concern,
                 collation=collation,
                 session=session,
+                retryable_write=retryable_write,
                 user_fields=_FIND_AND_MODIFY_DOC_FIELDS,
             )
             _check_write_command_response(out)
 
             return out.get("value")
 
         return await self._database.client._retryable_write(
+            write_concern.acknowledged,
             _find_and_modify_helper,
             session,
             operation=_Op.FIND_AND_MODIFY,