PYTHON-3771 add types to client_options.py (#1266)

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.