From 0a94c0a547ec743beabb69e7ceedcc2789afa907 Mon Sep 17 00:00:00 2001 From: Noah Stapp Date: Mon, 31 Aug 2026 10:23:33 -0400 Subject: [PATCH] PYTHON-6012 - Consolidate TopologySettings into a single shared class --- pymongo/asynchronous/settings.py | 146 +++++--------------------- pymongo/settings_shared.py | 174 +++++++++++++++++++++++++++++++ pymongo/synchronous/settings.py | 146 +++++--------------------- test/asynchronous/test_client.py | 3 +- test/test_client.py | 3 +- 5 files changed, 226 insertions(+), 246 deletions(-) create mode 100644 pymongo/settings_shared.py diff --git a/pymongo/asynchronous/settings.py b/pymongo/asynchronous/settings.py index e3c2ee7fb3..f14a11636f 100644 --- a/pymongo/asynchronous/settings.py +++ b/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, + ) diff --git a/pymongo/settings_shared.py b/pymongo/settings_shared.py new file mode 100644 index 0000000000..5008ddc335 --- /dev/null +++ b/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} diff --git a/pymongo/synchronous/settings.py b/pymongo/synchronous/settings.py index bc664e8421..6ac31ac339 100644 --- a/pymongo/synchronous/settings.py +++ b/pymongo/synchronous/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.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.settings_shared import _BaseTopologySettings from pymongo.synchronous import monitor, pool from pymongo.synchronous.pool import Pool -from pymongo.topology_description import TOPOLOGY_TYPE, _ServerSelector +from pymongo.topology_description import _ServerSelector _IS_SYNC = True -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, + ) diff --git a/test/asynchronous/test_client.py b/test/asynchronous/test_client.py index 92da29145c..bc00b54eac 100644 --- a/test/asynchronous/test_client.py +++ b/test/asynchronous/test_client.py @@ -67,7 +67,6 @@ from pymongo.asynchronous.pool import ( AsyncConnection, ) -from pymongo.asynchronous.settings import TOPOLOGY_TYPE from pymongo.asynchronous.topology import _ErrorContext from pymongo.client_options import ClientOptions from pymongo.common import _UUID_REPRESENTATIONS, CONNECT_TIMEOUT, MIN_SUPPORTED_WIRE_VERSION, has_c @@ -98,7 +97,7 @@ from pymongo.server_description import ServerDescription from pymongo.server_selectors import readable_server_selector, writable_server_selector from pymongo.server_type import SERVER_TYPE -from pymongo.topology_description import TopologyDescription +from pymongo.topology_description import TOPOLOGY_TYPE, TopologyDescription from pymongo.write_concern import WriteConcern from test.asynchronous import ( HAVE_IPADDRESS, diff --git a/test/test_client.py b/test/test_client.py index e0a3b3dcfe..7d3ab93ce1 100644 --- a/test/test_client.py +++ b/test/test_client.py @@ -95,9 +95,8 @@ from pymongo.synchronous.pool import ( Connection, ) -from pymongo.synchronous.settings import TOPOLOGY_TYPE from pymongo.synchronous.topology import _ErrorContext -from pymongo.topology_description import TopologyDescription +from pymongo.topology_description import TOPOLOGY_TYPE, TopologyDescription from pymongo.write_concern import WriteConcern from test import ( HAVE_IPADDRESS,