PYTHON-4993 - Reevaluate handling of asyncio.CancelledError

Author: NoahStapp

pymongo/_client_bulk_shared.py

@@ -16,6 +16,7 @@
 """Constants, types, and classes shared across Client Bulk Write API implementations."""
 from __future__ import annotations
 
+import asyncio
 from typing import TYPE_CHECKING, Any, Mapping, MutableMapping, NoReturn
 
 from pymongo.errors import ClientBulkWriteException, OperationFailure
@@ -75,5 +76,7 @@ def _throw_client_bulk_write_exception(
             )
             raise OperationFailure(errmsg, code, full_result)
     if isinstance(full_result["error"], BaseException):
+        if isinstance(full_result["error"], asyncio.CancelledError):
+            raise
         raise ClientBulkWriteException(full_result, verbose_results) from full_result["error"]
     raise ClientBulkWriteException(full_result, verbose_results)

pymongo/asynchronous/encryption.py

@@ -127,8 +127,6 @@ def _wrap_encryption_errors() -> Iterator[None]:
         # BSON encoding/decoding errors are unrelated to encryption so
         # we should propagate them unchanged.
         raise
-    except asyncio.CancelledError:
-        raise
     except Exception as exc:
         raise EncryptionError(exc) from exc
 
@@ -766,8 +764,6 @@ async def create_encrypted_collection(
                 await database.create_collection(name=name, **kwargs),
                 encrypted_fields,
             )
-        except asyncio.CancelledError:
-            raise
         except Exception as exc:
             raise EncryptedCollectionError(exc, encrypted_fields) from exc
 

pymongo/asynchronous/mongo_client.py

@@ -32,7 +32,6 @@
 """
 from __future__ import annotations
 
-import asyncio
 import contextlib
 import os
 import warnings
@@ -2038,8 +2037,6 @@ async def _process_kill_cursors(self) -> None:
         for address, cursor_id, conn_mgr in pinned_cursors:
             try:
                 await self._cleanup_cursor_lock(cursor_id, address, conn_mgr, None, False)
-            except asyncio.CancelledError:
-                raise
             except Exception as exc:
                 if isinstance(exc, InvalidOperation) and self._topology._closed:
                     # Raise the exception when client is closed so that it
@@ -2054,8 +2051,6 @@ async def _process_kill_cursors(self) -> None:
             for address, cursor_ids in address_to_cursor_ids.items():
                 try:
                     await self._kill_cursors(cursor_ids, address, topology, session=None)
-                except asyncio.CancelledError:
-                    raise
                 except Exception as exc:
                     if isinstance(exc, InvalidOperation) and self._topology._closed:
                         raise
@@ -2070,8 +2065,6 @@ async def _process_periodic_tasks(self) -> None:
         try:
             await self._process_kill_cursors()
             await self._topology.update_pool()
-        except asyncio.CancelledError:
-            raise
         except Exception as exc:
             if isinstance(exc, InvalidOperation) and self._topology._closed:
                 return

pymongo/asynchronous/monitor.py

@@ -16,7 +16,6 @@
 
 from __future__ import annotations
 
-import asyncio
 import atexit
 import logging
 import time
@@ -257,8 +256,6 @@ async def _check_server(self) -> ServerDescription:
                 details = cast(Mapping[str, Any], exc.details)
                 await self._topology.receive_cluster_time(details.get("$clusterTime"))
                 raise
-        except asyncio.CancelledError:
-            raise
         except ReferenceError:
             raise
         except Exception as error:
@@ -424,8 +421,6 @@ def _get_seedlist(self) -> Optional[list[tuple[str, Any]]]:
             if len(seedlist) == 0:
                 # As per the spec: this should be treated as a failure.
                 raise Exception
-        except asyncio.CancelledError:
-            raise
         except Exception:
             # As per the spec, upon encountering an error:
             # - An error must not be raised
@@ -489,8 +484,6 @@ async def _run(self) -> None:
         except ReferenceError:
             # Topology was garbage-collected.
             await self.close()
-        except asyncio.CancelledError:
-            raise
         except Exception:
             await self._pool.reset()
 

pymongo/asynchronous/pool.py

@@ -704,8 +704,6 @@ def _close_conn(self) -> None:
         # shutdown.
         try:
             self.conn.close()
-        except asyncio.CancelledError:
-            raise
         except Exception:  # noqa: S110
             pass
 

pymongo/synchronous/encryption.py

@@ -15,7 +15,6 @@
 """Support for explicit client-side field level encryption."""
 from __future__ import annotations
 
-import asyncio
 import contextlib
 import enum
 import socket
@@ -127,8 +126,6 @@ def _wrap_encryption_errors() -> Iterator[None]:
         # BSON encoding/decoding errors are unrelated to encryption so
         # we should propagate them unchanged.
         raise
-    except asyncio.CancelledError:
-        raise
     except Exception as exc:
         raise EncryptionError(exc) from exc
 
@@ -760,8 +757,6 @@ def create_encrypted_collection(
                 database.create_collection(name=name, **kwargs),
                 encrypted_fields,
             )
-        except asyncio.CancelledError:
-            raise
         except Exception as exc:
             raise EncryptedCollectionError(exc, encrypted_fields) from exc
 

pymongo/synchronous/mongo_client.py

@@ -32,7 +32,6 @@
 """
 from __future__ import annotations
 
-import asyncio
 import contextlib
 import os
 import warnings
@@ -2032,8 +2031,6 @@ def _process_kill_cursors(self) -> None:
         for address, cursor_id, conn_mgr in pinned_cursors:
             try:
                 self._cleanup_cursor_lock(cursor_id, address, conn_mgr, None, False)
-            except asyncio.CancelledError:
-                raise
             except Exception as exc:
                 if isinstance(exc, InvalidOperation) and self._topology._closed:
                     # Raise the exception when client is closed so that it
@@ -2048,8 +2045,6 @@ def _process_kill_cursors(self) -> None:
             for address, cursor_ids in address_to_cursor_ids.items():
                 try:
                     self._kill_cursors(cursor_ids, address, topology, session=None)
-                except asyncio.CancelledError:
-                    raise
                 except Exception as exc:
                     if isinstance(exc, InvalidOperation) and self._topology._closed:
                         raise
@@ -2064,8 +2059,6 @@ def _process_periodic_tasks(self) -> None:
         try:
             self._process_kill_cursors()
             self._topology.update_pool()
-        except asyncio.CancelledError:
-            raise
         except Exception as exc:
             if isinstance(exc, InvalidOperation) and self._topology._closed:
                 return

pymongo/synchronous/monitor.py

@@ -16,7 +16,6 @@
 
 from __future__ import annotations
 
-import asyncio
 import atexit
 import logging
 import time
@@ -257,8 +256,6 @@ def _check_server(self) -> ServerDescription:
                 details = cast(Mapping[str, Any], exc.details)
                 self._topology.receive_cluster_time(details.get("$clusterTime"))
                 raise
-        except asyncio.CancelledError:
-            raise
         except ReferenceError:
             raise
         except Exception as error:
@@ -424,8 +421,6 @@ def _get_seedlist(self) -> Optional[list[tuple[str, Any]]]:
             if len(seedlist) == 0:
                 # As per the spec: this should be treated as a failure.
                 raise Exception
-        except asyncio.CancelledError:
-            raise
         except Exception:
             # As per the spec, upon encountering an error:
             # - An error must not be raised
@@ -489,8 +484,6 @@ def _run(self) -> None:
         except ReferenceError:
             # Topology was garbage-collected.
             self.close()
-        except asyncio.CancelledError:
-            raise
         except Exception:
             self._pool.reset()
 

pymongo/synchronous/pool.py

@@ -702,8 +702,6 @@ def _close_conn(self) -> None:
         # shutdown.
         try:
             self.conn.close()
-        except asyncio.CancelledError:
-            raise
         except Exception:  # noqa: S110
             pass