From 75e56b26b8dac7b90dfd684644fe96130266e0a8 Mon Sep 17 00:00:00 2001 From: sleepyStick Date: Wed, 26 Jul 2023 11:40:12 -0700 Subject: [PATCH 1/3] PYTHON-3808 add types to monitor.py --- pymongo/monitor.py | 83 ++++++++++++++++++++++++----------------- pymongo/srv_resolver.py | 8 ++-- test/pymongo_mocks.py | 2 +- 3 files changed, 54 insertions(+), 39 deletions(-) diff --git a/pymongo/monitor.py b/pymongo/monitor.py index 2fc0bf8bab..b2376674fc 100644 --- a/pymongo/monitor.py +++ b/pymongo/monitor.py @@ -14,10 +14,12 @@ """Class to monitor a MongoDB server on a background thread.""" +from __future__ import annotations + import atexit import time import weakref -from typing import Any, Mapping, cast +from typing import TYPE_CHECKING, Any, List, Mapping, Optional, Tuple, cast from pymongo import common, periodic_executor from pymongo._csot import MovingMinimum @@ -29,8 +31,15 @@ from pymongo.server_description import ServerDescription from pymongo.srv_resolver import _SrvResolver +if TYPE_CHECKING: + from OpenSSL.SSL import Connection + + from pymongo.pool import Pool + from pymongo.settings import TopologySettings + from pymongo.topology import Topology + -def _sanitize(error): +def _sanitize(error: Exception) -> None: """PYTHON-2433 Clear error traceback info.""" error.__traceback__ = None error.__context__ = None @@ -38,7 +47,7 @@ def _sanitize(error): class MonitorBase: - def __init__(self, topology, name, interval, min_interval): + def __init__(self, topology: Topology, name: str, interval: int, min_interval: float): """Base class to do periodic work on a background thread. The background thread is signaled to stop when the Topology or @@ -46,7 +55,7 @@ def __init__(self, topology, name, interval, min_interval): """ # We strongly reference the executor and it weakly references us via # this closure. When the monitor is freed, stop the executor soon. - def target(): + def target() -> bool: monitor = self_ref() if monitor is None: return False # Stop the executor. @@ -59,7 +68,7 @@ def target(): self._executor = executor - def _on_topology_gc(dummy=None): + def _on_topology_gc(dummy: Optional[Topology] = None) -> None: # This prevents GC from waiting 10 seconds for hello to complete # See test_cleanup_executors_on_client_del. monitor = self_ref() @@ -71,35 +80,41 @@ def _on_topology_gc(dummy=None): self._topology = weakref.proxy(topology, _on_topology_gc) _register(self) - def open(self): + def open(self) -> None: """Start monitoring, or restart after a fork. Multiple calls have no effect. """ self._executor.open() - def gc_safe_close(self): + def gc_safe_close(self) -> None: """GC safe close.""" self._executor.close() - def close(self): + def close(self) -> None: """Close and stop monitoring. open() restarts the monitor after closing. """ self.gc_safe_close() - def join(self, timeout=None): + def join(self, timeout: Optional[int] = None) -> None: """Wait for the monitor to stop.""" self._executor.join(timeout) - def request_check(self): + def request_check(self) -> None: """If the monitor is sleeping, wake it soon.""" self._executor.wake() class Monitor(MonitorBase): - def __init__(self, server_description, topology, pool, topology_settings): + def __init__( + self, + server_description: ServerDescription, + topology: Topology, + pool: Pool, + topology_settings: TopologySettings, + ): """Class to monitor a MongoDB server on a background thread. Pass an initial ServerDescription, a Topology, a Pool, and @@ -128,7 +143,7 @@ def __init__(self, server_description, topology, pool, topology_settings): ) self.heartbeater = None - def cancel_check(self): + def cancel_check(self) -> None: """Cancel any concurrent hello check. Note: this is called from a weakref.proxy callback and MUST NOT take @@ -141,7 +156,7 @@ def cancel_check(self): # (depending on the platform). context.cancel() - def _start_rtt_monitor(self): + def _start_rtt_monitor(self) -> None: """Start an _RttMonitor that periodically runs ping.""" # If this monitor is closed directly before (or during) this open() # call, the _RttMonitor will not be closed. Checking if this monitor @@ -150,23 +165,23 @@ def _start_rtt_monitor(self): if self._executor._stopped: self._rtt_monitor.close() - def gc_safe_close(self): + def gc_safe_close(self) -> None: self._executor.close() self._rtt_monitor.gc_safe_close() self.cancel_check() - def close(self): + def close(self) -> None: self.gc_safe_close() self._rtt_monitor.close() # Increment the generation and maybe close the socket. If the executor # thread has the socket checked out, it will be closed when checked in. self._reset_connection() - def _reset_connection(self): + def _reset_connection(self) -> None: # Clear our pooled connection. self._pool.reset() - def _run(self): + def _run(self) -> None: try: prev_sd = self._server_description try: @@ -203,7 +218,7 @@ def _run(self): # Topology was garbage-collected. self.close() - def _check_server(self): + def _check_server(self) -> ServerDescription: """Call hello or read the next streaming response. Returns a ServerDescription. @@ -234,7 +249,7 @@ def _check_server(self): # Server type defaults to Unknown. return ServerDescription(address, error=error) - def _check_once(self): + def _check_once(self) -> ServerDescription: """A single attempt to call hello. Returns a ServerDescription, or raises an exception. @@ -259,7 +274,7 @@ def _check_once(self): ) return sd - def _check_with_socket(self, conn): + def _check_with_socket(self, conn: Connection) -> Tuple[Hello, float]: """Return (Hello, round_trip_time). Can raise ConnectionFailure or OperationFailure. @@ -283,7 +298,7 @@ def _check_with_socket(self, conn): class SrvMonitor(MonitorBase): - def __init__(self, topology, topology_settings): + def __init__(self, topology: Topology, topology_settings: TopologySettings): """Class to poll SRV records on a background thread. Pass a Topology and a TopologySettings. @@ -300,7 +315,7 @@ def __init__(self, topology, topology_settings): self._seedlist = self._settings._seeds self._fqdn = self._settings.fqdn - def _run(self): + def _run(self) -> None: seedlist = self._get_seedlist() if seedlist: self._seedlist = seedlist @@ -310,7 +325,7 @@ def _run(self): # Topology was garbage-collected. self.close() - def _get_seedlist(self): + def _get_seedlist(self) -> Optional[List[Tuple[str, Any]]]: """Poll SRV records for a seedlist. Returns a list of ServerDescriptions. @@ -338,7 +353,7 @@ def _get_seedlist(self): class _RttMonitor(MonitorBase): - def __init__(self, topology, topology_settings, pool): + def __init__(self, topology: Topology, topology_settings: TopologySettings, pool: Pool): """Maintain round trip times for a server. The Topology is weakly referenced. @@ -355,30 +370,30 @@ def __init__(self, topology, topology_settings, pool): self._moving_min = MovingMinimum() self._lock = _create_lock() - def close(self): + def close(self) -> None: self.gc_safe_close() # Increment the generation and maybe close the socket. If the executor # thread has the socket checked out, it will be closed when checked in. self._pool.reset() - def add_sample(self, sample): + def add_sample(self, sample: float) -> None: """Add a RTT sample.""" with self._lock: self._moving_average.add_sample(sample) self._moving_min.add_sample(sample) - def get(self): + def get(self) -> Tuple[Optional[float], float]: """Get the calculated average, or None if no samples yet and the min.""" with self._lock: return self._moving_average.get(), self._moving_min.get() - def reset(self): + def reset(self) -> None: """Reset the average RTT.""" with self._lock: self._moving_average.reset() self._moving_min.reset() - def _run(self): + def _run(self) -> None: try: # NOTE: This thread is only run when using the streaming # heartbeat protocol (MongoDB 4.4+). @@ -391,7 +406,7 @@ def _run(self): except Exception: self._pool.reset() - def _ping(self): + def _ping(self) -> float: """Run a "hello" command and return the RTT.""" with self._pool.get_socket() as sock_info: if self._executor._stopped: @@ -407,16 +422,16 @@ def _ping(self): _MONITORS = set() -def _register(monitor): +def _register(monitor: MonitorBase) -> None: ref = weakref.ref(monitor, _unregister) _MONITORS.add(ref) -def _unregister(monitor_ref): +def _unregister(monitor_ref: weakref.ReferenceType[MonitorBase]) -> None: _MONITORS.remove(monitor_ref) -def _shutdown_monitors(): +def _shutdown_monitors() -> None: if _MONITORS is None: return @@ -432,7 +447,7 @@ def _shutdown_monitors(): monitor = None -def _shutdown_resources(): +def _shutdown_resources() -> None: # _shutdown_monitors/_shutdown_executors may already be GC'd at shutdown. shutdown = _shutdown_monitors if shutdown: # type:ignore[truthy-function] diff --git a/pymongo/srv_resolver.py b/pymongo/srv_resolver.py index 67b781cf9c..4704c5a4cf 100644 --- a/pymongo/srv_resolver.py +++ b/pymongo/srv_resolver.py @@ -56,7 +56,7 @@ def _resolve(*args: Any, **kwargs: Any) -> resolver.Answer: class _SrvResolver: def __init__( self, - fqdn: str, + fqdn: Optional[str], connect_timeout: Optional[float], srv_service_name: str, srv_max_hosts: int = 0, @@ -67,13 +67,13 @@ def __init__( self.__srv_max_hosts = srv_max_hosts or 0 # Validate the fully qualified domain name. try: - ipaddress.ip_address(fqdn) + ipaddress.ip_address(fqdn) # type: ignore[arg-type] raise ConfigurationError(_INVALID_HOST_MSG % ("an IP address",)) except ValueError: pass try: - self.__plist = self.__fqdn.split(".")[1:] + self.__plist = self.__fqdn.split(".")[1:] # type: ignore[union-attr] except Exception: raise ConfigurationError(_INVALID_HOST_MSG % (fqdn,)) self.__slen = len(self.__plist) @@ -95,7 +95,7 @@ def get_options(self) -> Optional[str]: def _resolve_uri(self, encapsulate_errors: bool) -> resolver.Answer: try: results = _resolve( - "_" + self.__srv + "._tcp." + self.__fqdn, "SRV", lifetime=self.__connect_timeout + "_" + self.__srv + "._tcp." + self.__fqdn, "SRV", lifetime=self.__connect_timeout # type: ignore[operator] ) except Exception as exc: if not encapsulate_errors: diff --git a/test/pymongo_mocks.py b/test/pymongo_mocks.py index 2e7fda21e0..73e3bb4464 100644 --- a/test/pymongo_mocks.py +++ b/test/pymongo_mocks.py @@ -85,7 +85,7 @@ def __init__(self, client, server_description, topology, pool, topology_settings def _check_once(self): client = self.client address = self._server_description.address - response, rtt = client.mock_hello("%s:%d" % address) + response, rtt = client.mock_hello("%s:%d" % address) # type: ignore[str-format] return ServerDescription(address, Hello(response), rtt) From 65389a9aae7938e7bf412c4678f165c55a57bce4 Mon Sep 17 00:00:00 2001 From: sleepyStick Date: Mon, 31 Jul 2023 17:10:04 -0700 Subject: [PATCH 2/3] PYTHON-3808 fix fqdn type --- pymongo/monitor.py | 3 ++- pymongo/srv_resolver.py | 8 ++++---- 2 files changed, 6 insertions(+), 5 deletions(-) diff --git a/pymongo/monitor.py b/pymongo/monitor.py index b2376674fc..89ed7ec466 100644 --- a/pymongo/monitor.py +++ b/pymongo/monitor.py @@ -313,7 +313,8 @@ def __init__(self, topology: Topology, topology_settings: TopologySettings): ) self._settings = topology_settings self._seedlist = self._settings._seeds - self._fqdn = self._settings.fqdn + assert isinstance(self._settings.fqdn, str) + self._fqdn: str = self._settings.fqdn def _run(self) -> None: seedlist = self._get_seedlist() diff --git a/pymongo/srv_resolver.py b/pymongo/srv_resolver.py index 4704c5a4cf..67b781cf9c 100644 --- a/pymongo/srv_resolver.py +++ b/pymongo/srv_resolver.py @@ -56,7 +56,7 @@ def _resolve(*args: Any, **kwargs: Any) -> resolver.Answer: class _SrvResolver: def __init__( self, - fqdn: Optional[str], + fqdn: str, connect_timeout: Optional[float], srv_service_name: str, srv_max_hosts: int = 0, @@ -67,13 +67,13 @@ def __init__( self.__srv_max_hosts = srv_max_hosts or 0 # Validate the fully qualified domain name. try: - ipaddress.ip_address(fqdn) # type: ignore[arg-type] + ipaddress.ip_address(fqdn) raise ConfigurationError(_INVALID_HOST_MSG % ("an IP address",)) except ValueError: pass try: - self.__plist = self.__fqdn.split(".")[1:] # type: ignore[union-attr] + self.__plist = self.__fqdn.split(".")[1:] except Exception: raise ConfigurationError(_INVALID_HOST_MSG % (fqdn,)) self.__slen = len(self.__plist) @@ -95,7 +95,7 @@ def get_options(self) -> Optional[str]: def _resolve_uri(self, encapsulate_errors: bool) -> resolver.Answer: try: results = _resolve( - "_" + self.__srv + "._tcp." + self.__fqdn, "SRV", lifetime=self.__connect_timeout # type: ignore[operator] + "_" + self.__srv + "._tcp." + self.__fqdn, "SRV", lifetime=self.__connect_timeout ) except Exception as exc: if not encapsulate_errors: From 4d7dcacb4ea62492be01cfb5011c4e3320a474ec Mon Sep 17 00:00:00 2001 From: sleepyStick Date: Mon, 31 Jul 2023 18:41:58 -0700 Subject: [PATCH 3/3] PYTHON-3808 fix connection import --- pymongo/monitor.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/pymongo/monitor.py b/pymongo/monitor.py index 8568de0284..3fb20d0b19 100644 --- a/pymongo/monitor.py +++ b/pymongo/monitor.py @@ -32,9 +32,7 @@ from pymongo.srv_resolver import _SrvResolver if TYPE_CHECKING: - from OpenSSL.SSL import Connection - - from pymongo.pool import Pool + from pymongo.pool import Connection, Pool from pymongo.settings import TopologySettings from pymongo.topology import Topology