diff --git a/pymongo/client_options.py b/pymongo/client_options.py index 2e39b843e..7e5be6928 100644 --- a/pymongo/client_options.py +++ b/pymongo/client_options.py @@ -13,25 +13,38 @@ # permissions and limitations under the License. """Tools to parse mongo client options.""" +from __future__ import annotations -from typing import Optional +from typing import TYPE_CHECKING, Any, Callable, Mapping, Optional, Tuple from bson.codec_options import _parse_codec_options from pymongo import common -from pymongo.auth import _build_credentials_tuple +from pymongo.auth import MongoCredential, _build_credentials_tuple from pymongo.common import validate_boolean from pymongo.compression_support import CompressionSettings from pymongo.errors import ConfigurationError from pymongo.monitoring import _EventListeners from pymongo.pool import PoolOptions from pymongo.read_concern import ReadConcern -from pymongo.read_preferences import make_read_preference, read_pref_mode_from_name +from pymongo.read_preferences import ( + _ServerMode, + make_read_preference, + read_pref_mode_from_name, +) from pymongo.server_selectors import any_server_selector from pymongo.ssl_support import get_ssl_context from pymongo.write_concern import WriteConcern +if TYPE_CHECKING: + from bson.codec_options import CodecOptions + from pymongo.encryption import AutoEncryptionOpts + from pymongo.pyopenssl_context import SSLContext + from pymongo.server_selectors import Selection -def _parse_credentials(username, password, database, options): + +def _parse_credentials( + username: str, password: str, database: Optional[str], options: Mapping[str, Any] +) -> Optional[MongoCredential]: """Parse authentication credentials.""" mechanism = options.get("authmechanism", "DEFAULT" if username else None) source = options.get("authsource") @@ -40,7 +53,7 @@ def _parse_credentials(username, password, database, options): return None -def _parse_read_preference(options): +def _parse_read_preference(options: Mapping[str, Any]) -> _ServerMode: """Parse read preference options.""" if "read_preference" in options: return options["read_preference"] @@ -52,7 +65,7 @@ def _parse_read_preference(options): return make_read_preference(mode, tags, max_staleness) -def _parse_write_concern(options): +def _parse_write_concern(options: Mapping[str, Any]) -> WriteConcern: """Parse write concern options.""" concern = options.get("w") wtimeout = options.get("wtimeoutms") @@ -61,13 +74,13 @@ def _parse_write_concern(options): return WriteConcern(concern, wtimeout, j, fsync) -def _parse_read_concern(options): +def _parse_read_concern(options: Mapping[str, Any]) -> ReadConcern: """Parse read concern options.""" concern = options.get("readconcernlevel") return ReadConcern(concern) -def _parse_ssl_options(options): +def _parse_ssl_options(options: Mapping[str, Any]) -> Tuple[Optional[SSLContext], bool]: """Parse ssl options.""" use_tls = options.get("tls") if use_tls is not None: @@ -126,7 +139,9 @@ def _parse_ssl_options(options): return None, allow_invalid_hostnames -def _parse_pool_options(username, password, database, options): +def _parse_pool_options( + username: str, password: str, database: Optional[str], options: Mapping[str, Any] +) -> PoolOptions: """Parse connection pool options.""" credentials = _parse_credentials(username, password, database, options) max_pool_size = options.get("maxpoolsize", common.MAX_POOL_SIZE) @@ -175,7 +190,9 @@ class ClientOptions: instead. """ - def __init__(self, username, password, database, options): + def __init__( + self, username: str, password: str, database: Optional[str], options: Mapping[str, Any] + ): self.__options = options self.__codec_options = _parse_codec_options(options) self.__direct_connection = options.get("directconnection") @@ -200,66 +217,66 @@ class ClientOptions: self.__timeout = options.get("timeoutms") @property - def _options(self): + def _options(self) -> Mapping[str, Any]: """The original options used to create this ClientOptions.""" return self.__options @property - def connect(self): + def connect(self) -> Optional[bool]: """Whether to begin discovering a MongoDB topology automatically.""" return self.__connect @property - def codec_options(self): + def codec_options(self) -> CodecOptions: """A :class:`~bson.codec_options.CodecOptions` instance.""" return self.__codec_options @property - def direct_connection(self): + def direct_connection(self) -> Optional[bool]: """Whether to connect to the deployment in 'Single' topology.""" return self.__direct_connection @property - def local_threshold_ms(self): + def local_threshold_ms(self) -> int: """The local threshold for this instance.""" return self.__local_threshold_ms @property - def server_selection_timeout(self): + def server_selection_timeout(self) -> int: """The server selection timeout for this instance in seconds.""" return self.__server_selection_timeout @property - def server_selector(self): + def server_selector(self) -> Callable[[Selection], Selection]: return self.__server_selector @property - def heartbeat_frequency(self): + def heartbeat_frequency(self) -> int: """The monitoring frequency in seconds.""" return self.__heartbeat_frequency @property - def pool_options(self): + def pool_options(self) -> PoolOptions: """A :class:`~pymongo.pool.PoolOptions` instance.""" return self.__pool_options @property - def read_preference(self): + def read_preference(self) -> _ServerMode: """A read preference instance.""" return self.__read_preference @property - def replica_set_name(self): + def replica_set_name(self) -> Optional[str]: """Replica set name or None.""" return self.__replica_set_name @property - def write_concern(self): + def write_concern(self) -> WriteConcern: """A :class:`~pymongo.write_concern.WriteConcern` instance.""" return self.__write_concern @property - def read_concern(self): + def read_concern(self) -> ReadConcern: """A :class:`~pymongo.read_concern.ReadConcern` instance.""" return self.__read_concern @@ -272,27 +289,27 @@ class ClientOptions: return self.__timeout @property - def retry_writes(self): + def retry_writes(self) -> bool: """If this instance should retry supported write operations.""" return self.__retry_writes @property - def retry_reads(self): + def retry_reads(self) -> bool: """If this instance should retry supported read operations.""" return self.__retry_reads @property - def auto_encryption_opts(self): + def auto_encryption_opts(self) -> Optional[AutoEncryptionOpts]: """A :class:`~pymongo.encryption.AutoEncryptionOpts` or None.""" return self.__auto_encryption_opts @property - def load_balanced(self): + def load_balanced(self) -> Optional[bool]: """True if the client was configured to connect to a load balancer.""" return self.__load_balanced @property - def event_listeners(self): + def event_listeners(self) -> _EventListeners: """The event listeners registered for this client. See :mod:`~pymongo.monitoring` for details.