PYTHON-6012 - Consolidate TopologySettings into a single shared class

pymongo/asynchronous/settings.py

@@ -17,24 +17,22 @@
 from __future__ import annotations
 
 import threading
-import traceback
 from collections.abc import Collection
-from typing import Any, Optional, Union
+from typing import Optional
 
 from bson.objectid import ObjectId
 from pymongo import common
 from pymongo.asynchronous import monitor, pool
 from pymongo.asynchronous.pool import Pool
 from pymongo.common import LOCAL_THRESHOLD_MS, SERVER_SELECTION_TIMEOUT
-from pymongo.errors import ConfigurationError
 from pymongo.pool_options import PoolOptions
-from pymongo.server_description import ServerDescription
-from pymongo.topology_description import TOPOLOGY_TYPE, _ServerSelector
+from pymongo.settings_shared import _BaseTopologySettings
+from pymongo.topology_description import _ServerSelector
 
 _IS_SYNC = False
 
 
-class TopologySettings:
+class TopologySettings(_BaseTopologySettings[type[Pool], type[monitor.Monitor]]):
     def __init__(
         self,
         seeds: Optional[Collection[tuple[str, int]]] = None,
@@ -59,118 +57,24 @@ def __init__(
 
         Take a list of (host, port) pairs and optional replica set name.
         """
-        if heartbeat_frequency < common.MIN_HEARTBEAT_INTERVAL:
-            raise ConfigurationError(
-                f"heartbeatFrequencyMS cannot be less than {common.MIN_HEARTBEAT_INTERVAL * 1000}"
-            )
-
-        self._seeds: Collection[tuple[str, int]] = seeds or [("localhost", 27017)]
-        self._replica_set_name = replica_set_name
-        self._pool_class: type[Pool] = pool_class or pool.Pool
-        self._pool_options: PoolOptions = pool_options or PoolOptions()
-        self._monitor_class: type[monitor.Monitor] = monitor_class or monitor.Monitor
-        self._condition_class: type[threading.Condition] = condition_class or threading.Condition
-        self._local_threshold_ms = local_threshold_ms
-        self._server_selection_timeout = server_selection_timeout
-        self._server_selector = server_selector
-        self._fqdn = fqdn
-        self._heartbeat_frequency = heartbeat_frequency
-        self._direct = direct_connection
-        self._load_balanced = load_balanced
-        self._srv_service_name = srv_service_name
-        self._srv_max_hosts = srv_max_hosts or 0
-        self._server_monitoring_mode = server_monitoring_mode
-        if topology_id is not None:
-            self._topology_id = topology_id
-        else:
-            self._topology_id = ObjectId()
-        # Store the allocation traceback to catch unclosed clients in the
-        # test suite.
-        self._stack = "".join(traceback.format_stack()[:-2])
-
-    @property
-    def seeds(self) -> Collection[tuple[str, int]]:
-        """List of server addresses."""
-        return self._seeds
-
-    @property
-    def replica_set_name(self) -> Optional[str]:
-        return self._replica_set_name
-
-    @property
-    def pool_class(self) -> type[Pool]:
-        return self._pool_class
-
-    @property
-    def pool_options(self) -> PoolOptions:
-        return self._pool_options
-
-    @property
-    def monitor_class(self) -> type[monitor.Monitor]:
-        return self._monitor_class
-
-    @property
-    def condition_class(self) -> type[threading.Condition]:
-        return self._condition_class
-
-    @property
-    def local_threshold_ms(self) -> int:
-        return self._local_threshold_ms
-
-    @property
-    def server_selection_timeout(self) -> int:
-        return self._server_selection_timeout
-
-    @property
-    def server_selector(self) -> Optional[_ServerSelector]:
-        return self._server_selector
-
-    @property
-    def heartbeat_frequency(self) -> int:
-        return self._heartbeat_frequency
-
-    @property
-    def fqdn(self) -> Optional[str]:
-        return self._fqdn
-
-    @property
-    def direct(self) -> Optional[bool]:
-        """Connect directly to a single server, or use a set of servers?
-
-        True if there is one seed and no replica_set_name.
-        """
-        return self._direct
-
-    @property
-    def load_balanced(self) -> Optional[bool]:
-        """True if the client was configured to connect to a load balancer."""
-        return self._load_balanced
-
-    @property
-    def srv_service_name(self) -> str:
-        """The srvServiceName."""
-        return self._srv_service_name
-
-    @property
-    def srv_max_hosts(self) -> int:
-        """The srvMaxHosts."""
-        return self._srv_max_hosts
-
-    @property
-    def server_monitoring_mode(self) -> str:
-        """The serverMonitoringMode."""
-        return self._server_monitoring_mode
-
-    def get_topology_type(self) -> int:
-        if self.load_balanced:
-            return TOPOLOGY_TYPE.LoadBalanced
-        elif self.direct:
-            return TOPOLOGY_TYPE.Single
-        elif self.replica_set_name is not None:
-            return TOPOLOGY_TYPE.ReplicaSetNoPrimary
-        else:
-            return TOPOLOGY_TYPE.Unknown
-
-    def get_server_descriptions(self) -> dict[Union[tuple[str, int], Any], ServerDescription]:
-        """Initial dict of (address, ServerDescription) for all seeds."""
-        return {address: ServerDescription(address) for address in self.seeds}
+        pool_class = pool_class or pool.Pool
+        monitor_class = monitor_class or monitor.Monitor
+        super().__init__(
+            pool_class,
+            monitor_class,
+            seeds,
+            replica_set_name,
+            pool_options,
+            condition_class,
+            local_threshold_ms,
+            server_selection_timeout,
+            heartbeat_frequency,
+            server_selector,
+            fqdn,
+            direct_connection,
+            load_balanced,
+            srv_service_name,
+            srv_max_hosts,
+            server_monitoring_mode,
+            topology_id,
+        )

pymongo/settings_shared.py

@@ -0,0 +1,174 @@
+# Copyright 2014-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
+#
+# https://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.
+
+"""Represent MongoClient's configuration."""
+
+from __future__ import annotations
+
+import threading
+import traceback
+from collections.abc import Collection
+from typing import Any, Generic, Optional, TypeVar, Union
+
+from bson.objectid import ObjectId
+from pymongo import common
+from pymongo.common import LOCAL_THRESHOLD_MS, SERVER_SELECTION_TIMEOUT
+from pymongo.errors import ConfigurationError
+from pymongo.pool_options import PoolOptions
+from pymongo.server_description import ServerDescription
+from pymongo.topology_description import TOPOLOGY_TYPE, _ServerSelector
+
+_PoolClassType = TypeVar("_PoolClassType")
+_MonitorClassType = TypeVar("_MonitorClassType")
+
+
+class _BaseTopologySettings(Generic[_PoolClassType, _MonitorClassType]):
+    def __init__(
+        self,
+        pool_class: _PoolClassType,
+        monitor_class: _MonitorClassType,
+        seeds: Optional[Collection[tuple[str, int]]] = None,
+        replica_set_name: Optional[str] = None,
+        pool_options: Optional[PoolOptions] = None,
+        condition_class: Optional[type[threading.Condition]] = None,
+        local_threshold_ms: int = LOCAL_THRESHOLD_MS,
+        server_selection_timeout: int = SERVER_SELECTION_TIMEOUT,
+        heartbeat_frequency: int = common.HEARTBEAT_FREQUENCY,
+        server_selector: Optional[_ServerSelector] = None,
+        fqdn: Optional[str] = None,
+        direct_connection: Optional[bool] = False,
+        load_balanced: Optional[bool] = None,
+        srv_service_name: str = common.SRV_SERVICE_NAME,
+        srv_max_hosts: int = 0,
+        server_monitoring_mode: str = common.SERVER_MONITORING_MODE,
+        topology_id: Optional[ObjectId] = None,
+    ):
+        """Represent MongoClient's configuration.
+
+        Take a list of (host, port) pairs and optional replica set name.
+        """
+        if heartbeat_frequency < common.MIN_HEARTBEAT_INTERVAL:
+            raise ConfigurationError(
+                f"heartbeatFrequencyMS cannot be less than {common.MIN_HEARTBEAT_INTERVAL * 1000}"
+            )
+        self._pool_class: _PoolClassType = pool_class
+        self._monitor_class: _MonitorClassType = monitor_class
+        self._seeds: Collection[tuple[str, int]] = seeds or [("localhost", 27017)]
+        self._replica_set_name = replica_set_name
+        self._pool_options: PoolOptions = pool_options or PoolOptions()
+        self._condition_class: type[threading.Condition] = condition_class or threading.Condition
+        self._local_threshold_ms = local_threshold_ms
+        self._server_selection_timeout = server_selection_timeout
+        self._server_selector = server_selector
+        self._fqdn = fqdn
+        self._heartbeat_frequency = heartbeat_frequency
+        self._direct = direct_connection
+        self._load_balanced = load_balanced
+        self._srv_service_name = srv_service_name
+        self._srv_max_hosts = srv_max_hosts or 0
+        self._server_monitoring_mode = server_monitoring_mode
+        if topology_id is not None:
+            self._topology_id = topology_id
+        else:
+            self._topology_id = ObjectId()
+        # Store the allocation traceback to catch unclosed clients in the
+        # test suite.
+        self._stack = "".join(traceback.format_stack()[:-3])
+
+    @property
+    def seeds(self) -> Collection[tuple[str, int]]:
+        """List of server addresses."""
+        return self._seeds
+
+    @property
+    def replica_set_name(self) -> Optional[str]:
+        return self._replica_set_name
+
+    @property
+    def pool_class(self) -> _PoolClassType:
+        return self._pool_class
+
+    @property
+    def pool_options(self) -> PoolOptions:
+        return self._pool_options
+
+    @property
+    def monitor_class(self) -> _MonitorClassType:
+        return self._monitor_class
+
+    @property
+    def condition_class(self) -> type[threading.Condition]:
+        return self._condition_class
+
+    @property
+    def local_threshold_ms(self) -> int:
+        return self._local_threshold_ms
+
+    @property
+    def server_selection_timeout(self) -> int:
+        return self._server_selection_timeout
+
+    @property
+    def server_selector(self) -> Optional[_ServerSelector]:
+        return self._server_selector
+
+    @property
+    def heartbeat_frequency(self) -> int:
+        return self._heartbeat_frequency
+
+    @property
+    def fqdn(self) -> Optional[str]:
+        return self._fqdn
+
+    @property
+    def direct(self) -> Optional[bool]:
+        """Connect directly to a single server, or use a set of servers?
+
+        True if there is one seed and no replica_set_name.
+        """
+        return self._direct
+
+    @property
+    def load_balanced(self) -> Optional[bool]:
+        """True if the client was configured to connect to a load balancer."""
+        return self._load_balanced
+
+    @property
+    def srv_service_name(self) -> str:
+        """The srvServiceName."""
+        return self._srv_service_name
+
+    @property
+    def srv_max_hosts(self) -> int:
+        """The srvMaxHosts."""
+        return self._srv_max_hosts
+
+    @property
+    def server_monitoring_mode(self) -> str:
+        """The serverMonitoringMode."""
+        return self._server_monitoring_mode
+
+    def get_topology_type(self) -> int:
+        if self.load_balanced:
+            return TOPOLOGY_TYPE.LoadBalanced
+        elif self.direct:
+            return TOPOLOGY_TYPE.Single
+        elif self.replica_set_name is not None:
+            return TOPOLOGY_TYPE.ReplicaSetNoPrimary
+        else:
+            return TOPOLOGY_TYPE.Unknown
+
+    def get_server_descriptions(self) -> dict[Union[tuple[str, int], Any], ServerDescription]:
+        """Initial dict of (address, ServerDescription) for all seeds."""
+        return {address: ServerDescription(address) for address in self.seeds}