move init_background to only be called upon client connection

Author: sleepyStick

pymongo/asynchronous/mongo_client.py

@@ -859,9 +859,10 @@ def __init__(
             server_monitoring_mode=options.server_monitoring_mode,
         )
 
+        self._topology = Topology(self._topology_settings)
+
         self._opened = False
         self._closed = False
-        self._init_background()
 
         self._resolve_uri_info.update(
             {
@@ -878,17 +879,8 @@ def __init__(
             self._get_topology()  # type: ignore[unused-coroutine]
 
         self._encrypter = None
-        if self._options.auto_encryption_opts:
-            from pymongo.asynchronous.encryption import _Encrypter
-
-            self._encrypter = _Encrypter(self, self._options.auto_encryption_opts)
         self._timeout = self._options.timeout
 
-        if _HAS_REGISTER_AT_FORK:
-            # Add this client to the list of weakly referenced items.
-            # This will be used later if we fork.
-            AsyncMongoClient._clients[self._topology._topology_id] = self
-
     def _resolve_uri(self) -> None:
         keyword_opts = self._resolve_uri_info["keyword_opts"]
         seeds = set()
@@ -985,6 +977,17 @@ def _resolve_uri(self) -> None:
 
             self._topology = Topology(self._topology_settings)
 
+            if self._options.auto_encryption_opts:
+                from pymongo.asynchronous.encryption import _Encrypter
+
+                self._encrypter = _Encrypter(self, self._options.auto_encryption_opts)
+            self._timeout = self._options.timeout
+
+            if _HAS_REGISTER_AT_FORK:
+                # Add this client to the list of weakly referenced items.
+                # This will be used later if we fork.
+                AsyncMongoClient._clients[self._topology._topology_id] = self
+
     def _normalize_and_validate_options(
         self, opts: common._CaseInsensitiveDictionary, seeds: set[tuple[str, int | None]]
     ) -> common._CaseInsensitiveDictionary:
@@ -1672,19 +1675,20 @@ async def close(self) -> None:
             await self._end_sessions(session_ids)
         # Stop the periodic task thread and then send pending killCursor
         # requests before closing the topology.
-        self._kill_cursors_executor.close()
-        await self._process_kill_cursors()
-        await self._topology.close()
-        if self._encrypter:
-            # TODO: PYTHON-1921 Encrypted MongoClients cannot be re-opened.
-            await self._encrypter.close()
-        self._closed = True
-        if not _IS_SYNC:
-            await asyncio.gather(
-                self._topology.cleanup_monitors(),  # type: ignore[func-returns-value]
-                self._kill_cursors_executor.join(),  # type: ignore[func-returns-value]
-                return_exceptions=True,
-            )
+        if self._opened:
+            self._kill_cursors_executor.close()
+            await self._process_kill_cursors()
+            await self._topology.close()
+            if self._encrypter:
+                # TODO: PYTHON-1921 Encrypted MongoClients cannot be re-opened.
+                await self._encrypter.close()
+            self._closed = True
+            if not _IS_SYNC:
+                await asyncio.gather(
+                    self._topology.cleanup_monitors(),  # type: ignore[func-returns-value]
+                    self._kill_cursors_executor.join(),  # type: ignore[func-returns-value]
+                    return_exceptions=True,
+                )
 
     if not _IS_SYNC:
         # Add support for contextlib.aclosing.
@@ -1698,6 +1702,7 @@ async def _get_topology(self) -> Topology:
         """
         if not self._opened:
             self._resolve_uri()
+            self._init_background()
             await self._topology.open()
             async with self._lock:
                 self._kill_cursors_executor.open()

pymongo/synchronous/mongo_client.py

@@ -857,9 +857,10 @@ def __init__(
             server_monitoring_mode=options.server_monitoring_mode,
         )
 
+        self._topology = Topology(self._topology_settings)
+
         self._opened = False
         self._closed = False
-        self._init_background()
 
         self._resolve_uri_info.update(
             {
@@ -876,17 +877,8 @@ def __init__(
             self._get_topology()  # type: ignore[unused-coroutine]
 
         self._encrypter = None
-        if self._options.auto_encryption_opts:
-            from pymongo.synchronous.encryption import _Encrypter
-
-            self._encrypter = _Encrypter(self, self._options.auto_encryption_opts)
         self._timeout = self._options.timeout
 
-        if _HAS_REGISTER_AT_FORK:
-            # Add this client to the list of weakly referenced items.
-            # This will be used later if we fork.
-            MongoClient._clients[self._topology._topology_id] = self
-
     def _resolve_uri(self) -> None:
         keyword_opts = self._resolve_uri_info["keyword_opts"]
         seeds = set()
@@ -983,6 +975,17 @@ def _resolve_uri(self) -> None:
 
             self._topology = Topology(self._topology_settings)
 
+            if self._options.auto_encryption_opts:
+                from pymongo.synchronous.encryption import _Encrypter
+
+                self._encrypter = _Encrypter(self, self._options.auto_encryption_opts)
+            self._timeout = self._options.timeout
+
+            if _HAS_REGISTER_AT_FORK:
+                # Add this client to the list of weakly referenced items.
+                # This will be used later if we fork.
+                MongoClient._clients[self._topology._topology_id] = self
+
     def _normalize_and_validate_options(
         self, opts: common._CaseInsensitiveDictionary, seeds: set[tuple[str, int | None]]
     ) -> common._CaseInsensitiveDictionary:
@@ -1666,19 +1669,20 @@ def close(self) -> None:
             self._end_sessions(session_ids)
         # Stop the periodic task thread and then send pending killCursor
         # requests before closing the topology.
-        self._kill_cursors_executor.close()
-        self._process_kill_cursors()
-        self._topology.close()
-        if self._encrypter:
-            # TODO: PYTHON-1921 Encrypted MongoClients cannot be re-opened.
-            self._encrypter.close()
-        self._closed = True
-        if not _IS_SYNC:
-            asyncio.gather(
-                self._topology.cleanup_monitors(),  # type: ignore[func-returns-value]
-                self._kill_cursors_executor.join(),  # type: ignore[func-returns-value]
-                return_exceptions=True,
-            )
+        if self._opened:
+            self._kill_cursors_executor.close()
+            self._process_kill_cursors()
+            self._topology.close()
+            if self._encrypter:
+                # TODO: PYTHON-1921 Encrypted MongoClients cannot be re-opened.
+                self._encrypter.close()
+            self._closed = True
+            if not _IS_SYNC:
+                asyncio.gather(
+                    self._topology.cleanup_monitors(),  # type: ignore[func-returns-value]
+                    self._kill_cursors_executor.join(),  # type: ignore[func-returns-value]
+                    return_exceptions=True,
+                )
 
     if not _IS_SYNC:
         # Add support for contextlib.closing.
@@ -1692,6 +1696,7 @@ def _get_topology(self) -> Topology:
         """
         if not self._opened:
             self._resolve_uri()
+            self._init_background()
             self._topology.open()
             with self._lock:
                 self._kill_cursors_executor.open()

test/asynchronous/test_client.py

@@ -1058,12 +1058,8 @@ async def test_uri_connect_option(self):
         self.assertFalse(client._topology._opened)
 
         # Ensure kill cursors thread has not been started.
-        if _IS_SYNC:
-            kc_thread = client._kill_cursors_executor._thread
-            self.assertFalse(kc_thread and kc_thread.is_alive())
-        else:
-            kc_task = client._kill_cursors_executor._task
-            self.assertFalse(kc_task and not kc_task.done())
+        # _kill_cursors_executor is initialized upon client connection
+        self.assertFalse(hasattr(client, "_kill_cursors_executor"))
         # Using the client should open topology and start the thread.
         await client.admin.command("ping")
         self.assertTrue(client._topology._opened)

test/test_client.py

@@ -1031,12 +1031,8 @@ def test_uri_connect_option(self):
         self.assertFalse(client._topology._opened)
 
         # Ensure kill cursors thread has not been started.
-        if _IS_SYNC:
-            kc_thread = client._kill_cursors_executor._thread
-            self.assertFalse(kc_thread and kc_thread.is_alive())
-        else:
-            kc_task = client._kill_cursors_executor._task
-            self.assertFalse(kc_task and not kc_task.done())
+        # _kill_cursors_executor is initialized upon client connection
+        self.assertFalse(hasattr(client, "_kill_cursors_executor"))
         # Using the client should open topology and start the thread.
         client.admin.command("ping")
         self.assertTrue(client._topology._opened)