PYTHON-4724 - Prohibit AsyncMongoClient from being used across multiple event loops (#2256)

pymongo/asynchronous/mongo_client.py

@@ -878,6 +878,7 @@ class AsyncMongoClient(common.BaseObject, Generic[_DocumentType]):
 
         self._opened = False
         self._closed = False
+        self._loop: Optional[asyncio.AbstractEventLoop] = None
         if not is_srv:
             self._init_background()
 
@@ -1709,6 +1710,13 @@ class AsyncMongoClient(common.BaseObject, Generic[_DocumentType]):
         If this client was created with "connect=False", calling _get_topology
         launches the connection process in the background.
         """
+        if not _IS_SYNC:
+            if self._loop is None:
+                self._loop = asyncio.get_running_loop()
+            elif self._loop != asyncio.get_running_loop():
+                raise RuntimeError(
+                    "Cannot use AsyncMongoClient in different event loop. AsyncMongoClient uses low-level asyncio APIs that bind it to the event loop it was created on."
+                )
         if not self._opened:
             if self._resolve_srv_info["is_srv"]:
                 await self._resolve_srv()

pymongo/synchronous/mongo_client.py

@@ -876,6 +876,7 @@ class MongoClient(common.BaseObject, Generic[_DocumentType]):
 
         self._opened = False
         self._closed = False
+        self._loop: Optional[asyncio.AbstractEventLoop] = None
         if not is_srv:
             self._init_background()
 
@@ -1703,6 +1704,13 @@ class MongoClient(common.BaseObject, Generic[_DocumentType]):
         If this client was created with "connect=False", calling _get_topology
         launches the connection process in the background.
         """
+        if not _IS_SYNC:
+            if self._loop is None:
+                self._loop = asyncio.get_running_loop()
+            elif self._loop != asyncio.get_running_loop():
+                raise RuntimeError(
+                    "Cannot use MongoClient in different event loop. MongoClient uses low-level asyncio APIs that bind it to the event loop it was created on."
+                )
         if not self._opened:
             if self._resolve_srv_info["is_srv"]:
                 self._resolve_srv()

pyproject.toml

@@ -117,6 +117,8 @@ filterwarnings = [
     "module:unclosed <ssl.SSLSocket:ResourceWarning",
     "module:unclosed <socket object:ResourceWarning",
     "module:unclosed transport:ResourceWarning",
+    # pytest-asyncio known issue: https://github.com/pytest-dev/pytest-asyncio/issues/724
+    "module:unclosed event loop:ResourceWarning",
     # https://github.com/eventlet/eventlet/issues/818
     "module:please use dns.resolver.Resolver.resolve:DeprecationWarning",
     # https://github.com/dateutil/dateutil/issues/1314

test/asynchronous/test_async_loop_safety.py

@@ -0,0 +1,36 @@
+# Copyright 2025-present 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.
+
+"""Test that the asynchronous API detects event loop changes and fails correctly."""
+from __future__ import annotations
+
+import asyncio
+import unittest
+
+from pymongo import AsyncMongoClient
+
+
+class TestClientLoopSafety(unittest.TestCase):
+    def test_client_errors_on_different_loop(self):
+        client = AsyncMongoClient()
+        loop1 = asyncio.new_event_loop()
+        loop1.run_until_complete(client.aconnect())
+        loop2 = asyncio.new_event_loop()
+        with self.assertRaisesRegex(
+            RuntimeError, "Cannot use AsyncMongoClient in different event loop"
+        ):
+            loop2.run_until_complete(client.aconnect())
+        loop1.run_until_complete(client.close())
+        loop1.close()
+        loop2.close()

tools/synchro.py

@@ -180,7 +180,12 @@ gridfs_files = [
 
 def async_only_test(f: str) -> bool:
     """Return True for async tests that should not be converted to sync."""
-    return f in ["test_locks.py", "test_concurrency.py", "test_async_cancellation.py"]
+    return f in [
+        "test_locks.py",
+        "test_concurrency.py",
+        "test_async_cancellation.py",
+        "test_async_loop_safety.py",
+    ]
 
 
 test_files = [