cleanup and typing

Author: blink1073

pymongo/asynchronous/srv_resolver.py

@@ -17,7 +17,6 @@
 
 import ipaddress
 import random
-from importlib.metadata import version
 from typing import TYPE_CHECKING, Any, Optional, Union
 
 from pymongo.common import CONNECT_TIMEOUT, check_for_min_version
@@ -33,8 +32,7 @@ def _have_dnspython() -> bool:
     try:
         import dns  # noqa: F401
 
-        dns_version = version("dnspython")
-        required_version, is_valid = check_for_min_version(dns_version, "dnspython")
+        dns_version, required_version, is_valid = check_for_min_version("dnspython")
         if not is_valid:
             raise RuntimeError(
                 f"pymongo requires dnspython>={required_version}, "

pymongo/common.py

@@ -20,7 +20,7 @@
 import warnings
 from collections import OrderedDict, abc
 from difflib import get_close_matches
-from importlib.metadata import requires
+from importlib.metadata import requires, version
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -1095,23 +1095,23 @@ def has_c() -> bool:
         return False
 
 
-class Version(tuple):
+class Version(tuple[int]):
     """A class that can be used to compare version strings."""
 
-    def __new__(cls, *version):
+    def __new__(cls, *version: int) -> Version:
         padded_version = cls._padded(version, 4)
         return super().__new__(cls, tuple(padded_version))
 
     @classmethod
-    def _padded(cls, iter, length, padding=0):
+    def _padded(cls, iter: Any, length: int, padding: int = 0) -> list[int]:
         as_list = list(iter)
         if len(as_list) < length:
             for _ in range(length - len(as_list)):
                 as_list.append(padding)
         return as_list
 
     @classmethod
-    def from_string(cls, version_string):
+    def from_string(cls, version_string: str) -> Version:
         mod = 0
         bump_patch_level = False
         if version_string.endswith("+"):
@@ -1147,28 +1147,33 @@ def from_string(cls, version_string):
         return Version(*version)
 
     @classmethod
-    def from_version_array(cls, version_array):
+    def from_version_array(cls, version_array: Any) -> Version:
         version = list(version_array)
         if version[-1] < 0:
             version[-1] = -1
         version = cls._padded(version, 3)
         return Version(*version)
 
-    def at_least(self, *other_version):
+    def at_least(self, *other_version: Any) -> bool:
         return self >= Version(*other_version)
 
-    def __str__(self):
+    def __str__(self) -> str:
         return ".".join(map(str, self))
 
 
-def check_for_min_version(package_version: str, package_name: str) -> tuple[str, bool]:
-    package_version = Version.from_string(package_version)
+def check_for_min_version(package_name: str) -> tuple[str, str, bool]:
+    """Test whether an installed package is of the desired version."""
+    package_version_str = version(package_name)
+    package_version = Version.from_string(package_version_str)
     # Dependency is expected to be in one of the forms:
     # "pymongocrypt<2.0.0,>=1.13.0; extra == 'encryption'"
     # 'dnspython<3.0.0,>=1.16.0'
     #
-    requirement = [i for i in requires("pymongo") if i.startswith(package_name)][0]  # noqa: RUF015
+    requirements = requires("pymongo")
+    assert requirements is not None
+    requirement = [i for i in requirements if i.startswith(package_name)][0]  # noqa: RUF015
     if ";" in requirement:
         requirement = requirement.split(";")[0]
     required_version = requirement[requirement.find(">=") + 2 :]
-    return required_version, package_version >= Version.from_string(required_version)
+    is_valid = package_version >= Version.from_string(required_version)
+    return package_version_str, required_version, is_valid

pymongo/encryption_options.py

@@ -23,7 +23,7 @@
 from pymongo.uri_parser_shared import _parse_kms_tls_options
 
 try:
-    from pymongocrypt import __version__ as pymongocrypt_version  # type:ignore[import-untyped]
+    import pymongocrypt  # type:ignore[import-untyped]  # noqa: F401
 
     # Check for pymongocrypt>=1.10.
     from pymongocrypt import synchronous as _  # noqa: F401
@@ -42,7 +42,7 @@
 
 def check_min_pymongocrypt() -> None:
     """Raise an appropriate error if the min pymongocrypt is not installed."""
-    required_version, is_valid = check_for_min_version(pymongocrypt_version, "pymongocrypt")
+    pymongocrypt_version, required_version, is_valid = check_for_min_version("pymongocrypt")
     if not is_valid:
         raise ConfigurationError(
             f"client side encryption requires pymongocrypt>={required_version}, "

pymongo/synchronous/srv_resolver.py

@@ -17,7 +17,6 @@
 
 import ipaddress
 import random
-from importlib.metadata import version
 from typing import TYPE_CHECKING, Any, Optional, Union
 
 from pymongo.common import CONNECT_TIMEOUT, check_for_min_version
@@ -33,8 +32,7 @@ def _have_dnspython() -> bool:
     try:
         import dns  # noqa: F401
 
-        dns_version = version("dnspython")
-        required_version, is_valid = check_for_min_version(dns_version, "dnspython")
+        dns_version, required_version, is_valid = check_for_min_version("dnspython")
         if not is_valid:
             raise RuntimeError(
                 f"pymongo requires dnspython>={required_version}, "