address comments pt1

Author: sleepyStick

pymongo/asynchronous/client_session.py

@@ -996,7 +996,6 @@ def _txn_read_preference(self) -> Optional[_ServerMode]:
     def _materialize(self, logical_session_timeout_minutes: Optional[int] = None) -> None:
         if isinstance(self._server_session, _EmptyServerSession):
             old = self._server_session
-            assert self._client._topology is not None
             self._server_session = self._client._topology.get_server_session(
                 logical_session_timeout_minutes
             )

pymongo/asynchronous/mongo_client.py

@@ -757,7 +757,7 @@ def __init__(
             raise TypeError(f"port must be an instance of int, not {type(port)}")
         self._host = host
         self._port = port
-        self._topology: Optional[Topology] = None
+        self._topology: Topology = None  # type: ignore[assignment]
 
         # _pool_class, _monitor_class, and _condition_class are for deep
         # customization of PyMongo, e.g. Motor.
@@ -1036,7 +1036,6 @@ def _should_pin_cursor(self, session: Optional[AsyncClientSession]) -> Optional[
 
     def _after_fork(self) -> None:
         """Resets topology in a child after successfully forking."""
-        assert self._topology is not None
         self._init_background(self._topology._pid)
         # Reset the session pool to avoid duplicate sessions in the child process.
         self._topology._session_pool.reset()
@@ -1196,7 +1195,6 @@ def topology_description(self) -> TopologyDescription:
 
         .. versionadded:: 4.0
         """
-        assert self._topology is not None
         return self._topology.description
 
     @property
@@ -1210,7 +1208,6 @@ def nodes(self) -> FrozenSet[_Address]:
           to any servers, or a network partition causes it to lose connection
           to all servers.
         """
-        assert self._topology is not None
         description = self._topology.description
         return frozenset(s.address for s in description.known_servers)
 
@@ -1384,7 +1381,6 @@ def _ensure_session(
     def _send_cluster_time(
         self, command: MutableMapping[str, Any], session: Optional[AsyncClientSession]
     ) -> None:
-        assert self._topology is not None
         topology_time = self._topology.max_cluster_time()
         session_time = session.cluster_time if session else None
         if topology_time and session_time:
@@ -1570,7 +1566,6 @@ async def address(self) -> Optional[tuple[str, int]]:
 
         .. versionadded:: 3.0
         """
-        assert self._topology is not None
         topology_type = self._topology._description.topology_type
         if (
             topology_type == TOPOLOGY_TYPE.Sharded
@@ -1593,7 +1588,6 @@ async def primary(self) -> Optional[tuple[str, int]]:
         .. versionadded:: 3.0
            AsyncMongoClient gained this property in version 3.0.
         """
-        assert self._topology is not None
         return await self._topology.get_primary()  # type: ignore[return-value]
 
     @property
@@ -1607,7 +1601,6 @@ async def secondaries(self) -> set[_Address]:
         .. versionadded:: 3.0
            AsyncMongoClient gained this property in version 3.0.
         """
-        assert self._topology is not None
         return await self._topology.get_secondaries()
 
     @property
@@ -1618,7 +1611,6 @@ async def arbiters(self) -> set[_Address]:
         connected to a replica set, there are no arbiters, or this client was
         created without the `replicaSet` option.
         """
-        assert self._topology is not None
         return await self._topology.get_arbiters()
 
     @property
@@ -1677,25 +1669,26 @@ async def close(self) -> None:
         .. versionchanged:: 3.6
            End all server sessions created by this client.
         """
-        if self._topology is not None:
-            session_ids = self._topology.pop_all_sessions()
-            if session_ids:
-                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._topology is None:
+            return
+        session_ids = self._topology.pop_all_sessions()
+        if session_ids:
+            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 not _IS_SYNC:
         # Add support for contextlib.aclosing.
@@ -1711,12 +1704,10 @@ async def _get_topology(self) -> Topology:
             if self._resolve_srv_info["is_srv"]:
                 await self._resolve_srv()
                 self._init_background()
-            assert self._topology is not None
             await self._topology.open()
             async with self._lock:
                 self._kill_cursors_executor.open()
             self._opened = True
-        assert self._topology is not None
         return self._topology
 
     @contextlib.asynccontextmanager
@@ -1819,7 +1810,6 @@ async def _conn_from_server(
         # Thread safe: if the type is single it cannot change.
         # NOTE: We already opened the Topology when selecting a server so there's no need
         # to call _get_topology() again.
-        assert self._topology is not None
         single = self._topology.description.topology_type == TOPOLOGY_TYPE.Single
         async with self._checkout(server, session) as conn:
             if single:
@@ -2159,7 +2149,6 @@ async def _process_kill_cursors(self) -> None:
         """Process any pending kill cursors requests."""
         address_to_cursor_ids = defaultdict(list)
         pinned_cursors = []
-        assert self._topology is not None
 
         # Other threads or the GC may append to the queue concurrently.
         while True:
@@ -2201,7 +2190,6 @@ async def _process_periodic_tasks(self) -> None:
         """Process any pending kill cursors requests and
         maintain connection pool parameters.
         """
-        assert self._topology is not None
         try:
             await self._process_kill_cursors()
             await self._topology.update_pool()
@@ -2217,7 +2205,6 @@ def _return_server_session(
         """Internal: return a _ServerSession to the pool."""
         if isinstance(server_session, _EmptyServerSession):
             return None
-        assert self._topology is not None
         return self._topology.return_server_session(server_session)
 
     @contextlib.asynccontextmanager
@@ -2255,7 +2242,6 @@ async def _tmp_session(
     async def _process_response(
         self, reply: Mapping[str, Any], session: Optional[AsyncClientSession]
     ) -> None:
-        assert self._topology is not None
         await self._topology.receive_cluster_time(reply.get("$clusterTime"))
         if session is not None:
             session._process_response(reply)

pymongo/asynchronous/pool.py

@@ -1315,7 +1315,6 @@ async def connect(self, handler: Optional[_MongoClientErrorHandler] = None) -> A
             raise
 
         if handler:
-            assert handler.client._topology is not None
             await handler.client._topology.receive_cluster_time(conn._cluster_time)
 
         return conn

pymongo/synchronous/client_session.py

@@ -991,7 +991,6 @@ def _txn_read_preference(self) -> Optional[_ServerMode]:
     def _materialize(self, logical_session_timeout_minutes: Optional[int] = None) -> None:
         if isinstance(self._server_session, _EmptyServerSession):
             old = self._server_session
-            assert self._client._topology is not None
             self._server_session = self._client._topology.get_server_session(
                 logical_session_timeout_minutes
             )

pymongo/synchronous/mongo_client.py

@@ -755,7 +755,7 @@ def __init__(
             raise TypeError(f"port must be an instance of int, not {type(port)}")
         self._host = host
         self._port = port
-        self._topology: Optional[Topology] = None
+        self._topology: Topology = None  # type: ignore[assignment]
 
         # _pool_class, _monitor_class, and _condition_class are for deep
         # customization of PyMongo, e.g. Motor.
@@ -1034,7 +1034,6 @@ def _should_pin_cursor(self, session: Optional[ClientSession]) -> Optional[bool]
 
     def _after_fork(self) -> None:
         """Resets topology in a child after successfully forking."""
-        assert self._topology is not None
         self._init_background(self._topology._pid)
         # Reset the session pool to avoid duplicate sessions in the child process.
         self._topology._session_pool.reset()
@@ -1194,7 +1193,6 @@ def topology_description(self) -> TopologyDescription:
 
         .. versionadded:: 4.0
         """
-        assert self._topology is not None
         return self._topology.description
 
     @property
@@ -1208,7 +1206,6 @@ def nodes(self) -> FrozenSet[_Address]:
           to any servers, or a network partition causes it to lose connection
           to all servers.
         """
-        assert self._topology is not None
         description = self._topology.description
         return frozenset(s.address for s in description.known_servers)
 
@@ -1380,7 +1377,6 @@ def _ensure_session(self, session: Optional[ClientSession] = None) -> Optional[C
     def _send_cluster_time(
         self, command: MutableMapping[str, Any], session: Optional[ClientSession]
     ) -> None:
-        assert self._topology is not None
         topology_time = self._topology.max_cluster_time()
         session_time = session.cluster_time if session else None
         if topology_time and session_time:
@@ -1564,7 +1560,6 @@ def address(self) -> Optional[tuple[str, int]]:
 
         .. versionadded:: 3.0
         """
-        assert self._topology is not None
         topology_type = self._topology._description.topology_type
         if (
             topology_type == TOPOLOGY_TYPE.Sharded
@@ -1587,7 +1582,6 @@ def primary(self) -> Optional[tuple[str, int]]:
         .. versionadded:: 3.0
            MongoClient gained this property in version 3.0.
         """
-        assert self._topology is not None
         return self._topology.get_primary()  # type: ignore[return-value]
 
     @property
@@ -1601,7 +1595,6 @@ def secondaries(self) -> set[_Address]:
         .. versionadded:: 3.0
            MongoClient gained this property in version 3.0.
         """
-        assert self._topology is not None
         return self._topology.get_secondaries()
 
     @property
@@ -1612,7 +1605,6 @@ def arbiters(self) -> set[_Address]:
         connected to a replica set, there are no arbiters, or this client was
         created without the `replicaSet` option.
         """
-        assert self._topology is not None
         return self._topology.get_arbiters()
 
     @property
@@ -1671,25 +1663,26 @@ def close(self) -> None:
         .. versionchanged:: 3.6
            End all server sessions created by this client.
         """
-        if self._topology is not None:
-            session_ids = self._topology.pop_all_sessions()
-            if session_ids:
-                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._topology is None:
+            return
+        session_ids = self._topology.pop_all_sessions()
+        if session_ids:
+            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 not _IS_SYNC:
         # Add support for contextlib.closing.
@@ -1705,12 +1698,10 @@ def _get_topology(self) -> Topology:
             if self._resolve_srv_info["is_srv"]:
                 self._resolve_srv()
                 self._init_background()
-            assert self._topology is not None
             self._topology.open()
             with self._lock:
                 self._kill_cursors_executor.open()
             self._opened = True
-        assert self._topology is not None
         return self._topology
 
     @contextlib.contextmanager
@@ -1813,7 +1804,6 @@ def _conn_from_server(
         # Thread safe: if the type is single it cannot change.
         # NOTE: We already opened the Topology when selecting a server so there's no need
         # to call _get_topology() again.
-        assert self._topology is not None
         single = self._topology.description.topology_type == TOPOLOGY_TYPE.Single
         with self._checkout(server, session) as conn:
             if single:
@@ -2153,7 +2143,6 @@ def _process_kill_cursors(self) -> None:
         """Process any pending kill cursors requests."""
         address_to_cursor_ids = defaultdict(list)
         pinned_cursors = []
-        assert self._topology is not None
 
         # Other threads or the GC may append to the queue concurrently.
         while True:
@@ -2195,7 +2184,6 @@ def _process_periodic_tasks(self) -> None:
         """Process any pending kill cursors requests and
         maintain connection pool parameters.
         """
-        assert self._topology is not None
         try:
             self._process_kill_cursors()
             self._topology.update_pool()
@@ -2211,7 +2199,6 @@ def _return_server_session(
         """Internal: return a _ServerSession to the pool."""
         if isinstance(server_session, _EmptyServerSession):
             return None
-        assert self._topology is not None
         return self._topology.return_server_session(server_session)
 
     @contextlib.contextmanager
@@ -2247,7 +2234,6 @@ def _tmp_session(
             yield None
 
     def _process_response(self, reply: Mapping[str, Any], session: Optional[ClientSession]) -> None:
-        assert self._topology is not None
         self._topology.receive_cluster_time(reply.get("$clusterTime"))
         if session is not None:
             session._process_response(reply)

pymongo/synchronous/pool.py

@@ -1309,7 +1309,6 @@ def connect(self, handler: Optional[_MongoClientErrorHandler] = None) -> Connect
             raise
 
         if handler:
-            assert handler.client._topology is not None
             handler.client._topology.receive_cluster_time(conn._cluster_time)
 
         return conn