fix typing

Author: sleepyStick

pymongo/asynchronous/client_session.py

@@ -996,6 +996,7 @@ 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 = None
+        self._topology: Optional[Topology] = None
 
         # _pool_class, _monitor_class, and _condition_class are for deep
         # customization of PyMongo, e.g. Motor.
@@ -1036,6 +1036,7 @@ 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()
@@ -1195,6 +1196,7 @@ def topology_description(self) -> TopologyDescription:
 
         .. versionadded:: 4.0
         """
+        assert self._topology is not None
         return self._topology.description
 
     @property
@@ -1208,6 +1210,7 @@ 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)
 
@@ -1221,7 +1224,7 @@ def options(self) -> ClientOptions:
         """
         return self._options
 
-    def eq_props(self):
+    def eq_props(self) -> tuple[tuple[_Address, ...], Optional[str], Optional[str], str]:
         return (
             tuple(sorted(self._resolve_srv_info["seeds"])),
             self._options.replica_set_name,
@@ -1242,7 +1245,7 @@ def __ne__(self, other: Any) -> bool:
 
     def __hash__(self) -> int:
         if self._topology is None:
-            raise hash(self.eq_props())
+            return hash(self.eq_props())
         else:
             return hash(self._topology)
 
@@ -1387,6 +1390,7 @@ 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:
@@ -1572,6 +1576,7 @@ 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
@@ -1594,6 +1599,7 @@ 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,6 +1613,7 @@ 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
@@ -1617,6 +1624,7 @@ 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
@@ -1709,10 +1717,12 @@ 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
@@ -1815,6 +1825,7 @@ 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:
@@ -2154,6 +2165,7 @@ 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:
@@ -2195,6 +2207,7 @@ 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()
@@ -2210,6 +2223,7 @@ 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
@@ -2247,6 +2261,7 @@ 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)
@@ -2638,6 +2653,7 @@ async def handle(
             self.completed_handshake,
             self.service_id,
         )
+        assert self._client._topology is not None
         await self.client._topology.handle_error(self.server_address, err_ctx)
 
     async def __aenter__(self) -> _MongoClientErrorHandler:

pymongo/asynchronous/pool.py

@@ -1315,6 +1315,7 @@ 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,6 +991,7 @@ 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 = None
+        self._topology: Optional[Topology] = None
 
         # _pool_class, _monitor_class, and _condition_class are for deep
         # customization of PyMongo, e.g. Motor.
@@ -1034,6 +1034,7 @@ 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()
@@ -1193,6 +1194,7 @@ def topology_description(self) -> TopologyDescription:
 
         .. versionadded:: 4.0
         """
+        assert self._topology is not None
         return self._topology.description
 
     @property
@@ -1206,6 +1208,7 @@ 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)
 
@@ -1219,7 +1222,7 @@ def options(self) -> ClientOptions:
         """
         return self._options
 
-    def eq_props(self):
+    def eq_props(self) -> tuple[tuple[_Address, ...], Optional[str], Optional[str], str]:
         return (
             tuple(sorted(self._resolve_srv_info["seeds"])),
             self._options.replica_set_name,
@@ -1240,7 +1243,7 @@ def __ne__(self, other: Any) -> bool:
 
     def __hash__(self) -> int:
         if self._topology is None:
-            raise hash(self.eq_props())
+            return hash(self.eq_props())
         else:
             return hash(self._topology)
 
@@ -1383,6 +1386,7 @@ 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:
@@ -1566,6 +1570,7 @@ 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
@@ -1588,6 +1593,7 @@ 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,6 +1607,7 @@ 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
@@ -1611,6 +1618,7 @@ 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
@@ -1703,10 +1711,12 @@ 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
@@ -1809,6 +1819,7 @@ 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:
@@ -2148,6 +2159,7 @@ 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:
@@ -2189,6 +2201,7 @@ 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()
@@ -2204,6 +2217,7 @@ 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
@@ -2239,6 +2253,7 @@ 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)
@@ -2624,6 +2639,7 @@ def handle(
             self.completed_handshake,
             self.service_id,
         )
+        assert self._client._topology is not None
         self.client._topology.handle_error(self.server_address, err_ctx)
 
     def __enter__(self) -> _MongoClientErrorHandler:

pymongo/synchronous/pool.py

@@ -1309,6 +1309,7 @@ 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