♻️ refactor: improve code quality across package

Make specifier classes frozen kw-only dataclasses with from_string
factory classmethods instead of mutable __init__. This makes them
properly immutable and separates construction from parsing.

Also standardize module-level constants: private _LOGGER with
Final[type] annotations, compiled verbose regexes as module constants,
and deduplicate exe-checking logic into _check_exe/_is_new_exe helpers.

Author: gaborbernat

src/py_discovery/__init__.py

@@ -3,13 +3,14 @@
 from __future__ import annotations
 
 from importlib.metadata import version
+
 from ._cache import ContentStore, DiskCache, PyInfoCache
 from ._discovery import get_interpreter
 from ._py_info import PythonInfo
 from ._py_spec import PythonSpec
 
-
 __version__ = version("py-discovery")
+
 __all__ = [
     "ContentStore",
     "DiskCache",

src/py_discovery/_cached_py_info.py

@@ -16,7 +16,7 @@
 from pathlib import Path
 from shlex import quote
 from subprocess import Popen
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Final
 
 from ._cache import NoOpCache
 from ._py_info import PythonInfo
@@ -29,7 +29,7 @@
 
 _CACHE: OrderedDict[Path, PythonInfo | Exception] = OrderedDict()
 _CACHE[Path(sys.executable)] = PythonInfo()
-LOGGER = logging.getLogger(__name__)
+_LOGGER: Final[logging.Logger] = logging.getLogger(__name__)
 
 
 def from_exe(
@@ -45,7 +45,7 @@ def from_exe(
     if isinstance(result, Exception):
         if raise_on_error:
             raise result
-        LOGGER.info("%s", result)
+        _LOGGER.info("%s", result)
         result = None
     return result
 
@@ -105,7 +105,7 @@ def _get_via_file_cache(
         if py_info is None:
             failure, py_info = _run_subprocess(cls, exe, env)
             if failure is not None:
-                LOGGER.debug("first subprocess attempt failed for %s (%s), retrying", exe, failure)
+                _LOGGER.debug("first subprocess attempt failed for %s (%s), retrying", exe, failure)
                 failure, py_info = _run_subprocess(cls, exe, env)
             if failure is not None:
                 return failure
@@ -138,7 +138,7 @@ def _load_cached_py_info(
     return py_info
 
 
-COOKIE_LENGTH: int = 32
+COOKIE_LENGTH: Final[int] = 32
 
 
 def gen_cookie() -> str:
@@ -176,7 +176,7 @@ def _run_subprocess(
         env = dict(env)
         env.pop("__PYVENV_LAUNCHER__", None)
         env["PYTHONUTF8"] = "1"
-        LOGGER.debug("get interpreter info via cmd: %s", LogCmd(cmd))
+        _LOGGER.debug("get interpreter info via cmd: %s", LogCmd(cmd))
         try:
             process = Popen(
                 cmd,
@@ -220,7 +220,7 @@ def _run_subprocess(
             result = cls._from_json(out)
             result.executable = exe
         except json.JSONDecodeError as exc:
-            LOGGER.warning(
+            _LOGGER.warning(
                 "subprocess %s returned invalid JSON; raw stdout %d chars, start cookie %s, end cookie %s, "
                 "parsed output %d chars: %r",
                 exe,

src/py_discovery/_compat.py

@@ -5,18 +5,17 @@
 import functools
 import logging
 import os
-import sys
 import tempfile
+from typing import Final
 
-IS_WIN = sys.platform == "win32"
-LOGGER = logging.getLogger(__name__)
+_LOGGER: Final[logging.Logger] = logging.getLogger(__name__)
 
 
 @functools.lru_cache(maxsize=1)
 def fs_is_case_sensitive() -> bool:
     with tempfile.NamedTemporaryFile(prefix="TmP") as tmp_file:
         result = not os.path.exists(tmp_file.name.lower())
-    LOGGER.debug("filesystem is %scase-sensitive", "" if result else "not ")
+    _LOGGER.debug("filesystem is %scase-sensitive", "" if result else "not ")
     return result
 
 
@@ -25,7 +24,6 @@ def fs_path_id(path: str) -> str:
 
 
 __all__ = [
-    "IS_WIN",
     "fs_is_case_sensitive",
     "fs_path_id",
 ]

src/py_discovery/_discovery.py

@@ -5,11 +5,11 @@
 import sys
 from contextlib import suppress
 from pathlib import Path
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Final
 
 from platformdirs import user_data_path
 
-from ._compat import IS_WIN, fs_path_id
+from ._compat import fs_path_id
 from ._py_info import PythonInfo
 from ._py_spec import PythonSpec
 
@@ -18,7 +18,8 @@
 
     from ._cache import PyInfoCache
 
-LOGGER = logging.getLogger(__name__)
+_LOGGER: Final[logging.Logger] = logging.getLogger(__name__)
+IS_WIN: Final[bool] = sys.platform == "win32"
 
 
 def get_interpreter(
@@ -41,7 +42,7 @@ def _find_interpreter(
     env: Mapping[str, str] | None = None,
 ) -> PythonInfo | None:
     spec = PythonSpec.from_string_spec(key)
-    LOGGER.info("find interpreter for spec %r", spec)
+    _LOGGER.info("find interpreter for spec %r", spec)
     proposed_paths: set[tuple[str, bool]] = set()
     env = os.environ if env is None else env
     for interpreter, impl_must_match in propose_interpreters(spec, try_first_with, cache, env):
@@ -50,14 +51,37 @@ def _find_interpreter(
         proposed_key = interpreter.system_executable, impl_must_match
         if proposed_key in proposed_paths:
             continue
-        LOGGER.info("proposed %s", interpreter)
+        _LOGGER.info("proposed %s", interpreter)
         if interpreter.satisfies(spec, impl_must_match):
-            LOGGER.debug("accepted %s", interpreter)
+            _LOGGER.debug("accepted %s", interpreter)
             return interpreter
         proposed_paths.add(proposed_key)
     return None
 
 
+def _check_exe(path: str, tested_exes: set[str]) -> str | None:
+    """Resolve *path* to an absolute path and return it if not yet tested, otherwise ``None``."""
+    try:
+        os.lstat(path)
+    except OSError:
+        return None
+    exe_raw = os.path.abspath(path)
+    exe_id = fs_path_id(exe_raw)
+    if exe_id in tested_exes:
+        return None
+    tested_exes.add(exe_id)
+    return exe_raw
+
+
+def _is_new_exe(exe_raw: str, tested_exes: set[str]) -> bool:
+    """Return ``True`` and register *exe_raw* if it hasn't been tested yet."""
+    exe_id = fs_path_id(exe_raw)
+    if exe_id in tested_exes:
+        return False
+    tested_exes.add(exe_id)
+    return True
+
+
 def propose_interpreters(
     spec: PythonSpec,
     try_first_with: Iterable[str],
@@ -67,76 +91,41 @@ def propose_interpreters(
     env = os.environ if env is None else env
     tested_exes: set[str] = set()
     if spec.is_abs and spec.path is not None:
-        try:
-            os.lstat(spec.path)
-        except OSError:
-            pass
-        else:
-            exe_raw = os.path.abspath(spec.path)
-            exe_id = fs_path_id(exe_raw)
-            if exe_id not in tested_exes:  # pragma: no branch # first exe always new
-                tested_exes.add(exe_id)
-                yield PythonInfo.from_exe(exe_raw, cache, env=env), True
+        if exe_raw := _check_exe(spec.path, tested_exes):  # pragma: no branch # first exe always new
+            yield PythonInfo.from_exe(exe_raw, cache, env=env), True
         return
 
     for py_exe in try_first_with:
-        path = os.path.abspath(py_exe)
-        try:
-            os.lstat(path)
-        except OSError:
-            pass
-        else:
-            exe_raw = os.path.abspath(path)
-            exe_id = fs_path_id(exe_raw)
-            if exe_id in tested_exes:
-                continue
-            tested_exes.add(exe_id)
+        if exe_raw := _check_exe(os.path.abspath(py_exe), tested_exes):
             yield PythonInfo.from_exe(exe_raw, cache, env=env), True
 
     if spec.path is not None:
-        try:
-            os.lstat(spec.path)
-        except OSError:
-            pass
-        else:
-            exe_raw = os.path.abspath(spec.path)
-            exe_id = fs_path_id(exe_raw)
-            if exe_id not in tested_exes:  # pragma: no branch
-                tested_exes.add(exe_id)
-                yield PythonInfo.from_exe(exe_raw, cache, env=env), True
+        if exe_raw := _check_exe(spec.path, tested_exes):  # pragma: no branch
+            yield PythonInfo.from_exe(exe_raw, cache, env=env), True
         if spec.is_abs:  # pragma: no cover # relative spec.path is never abs
             return
     else:
         current_python = PythonInfo.current_system(cache)
-        exe_raw = str(current_python.executable)
-        exe_id = fs_path_id(exe_raw)
-        if exe_id not in tested_exes:
-            tested_exes.add(exe_id)
+        if _is_new_exe(str(current_python.executable), tested_exes):
             yield current_python, True
 
         if IS_WIN:  # pragma: win32 cover
             from ._windows import propose_interpreters
 
             for interpreter in propose_interpreters(spec, cache, env):
-                exe_raw = str(interpreter.executable)
-                exe_id = fs_path_id(exe_raw)
-                if exe_id in tested_exes:
-                    continue
-                tested_exes.add(exe_id)
-                yield interpreter, True
+                if _is_new_exe(str(interpreter.executable), tested_exes):
+                    yield interpreter, True
 
     find_candidates = path_exe_finder(spec)
     for pos, path in enumerate(get_paths(env)):
-        LOGGER.debug(LazyPathDump(pos, path, env))
+        _LOGGER.debug(LazyPathDump(pos, path, env))
         for exe, impl_must_match in find_candidates(path):
             exe_raw = str(exe)
             if resolved := _resolve_shim(exe_raw, env):
-                LOGGER.debug("resolved shim %s to %s", exe_raw, resolved)
+                _LOGGER.debug("resolved shim %s to %s", exe_raw, resolved)
                 exe_raw = resolved
-            exe_id = fs_path_id(exe_raw)
-            if exe_id in tested_exes:
+            if not _is_new_exe(exe_raw, tested_exes):
                 continue
-            tested_exes.add(exe_id)
             interpreter = PathPythonInfo.from_exe(exe_raw, cache, raise_on_error=False, env=env)
             if interpreter is not None:
                 yield interpreter, impl_must_match

src/py_discovery/_py_info.py

@@ -17,7 +17,7 @@
 import warnings
 from collections import OrderedDict, namedtuple
 from string import digits
-from typing import TYPE_CHECKING, ClassVar
+from typing import TYPE_CHECKING, ClassVar, Final
 
 if TYPE_CHECKING:
     from collections.abc import Mapping
@@ -26,15 +26,20 @@
     from ._py_spec import PythonSpec
 
 VersionInfo = namedtuple("VersionInfo", ["major", "minor", "micro", "releaselevel", "serial"])
-LOGGER = logging.getLogger(__name__)
+_LOGGER: Final[logging.Logger] = logging.getLogger(__name__)
 
 
 def _get_path_extensions():
     return list(OrderedDict.fromkeys(["", *os.environ.get("PATHEXT", "").lower().split(os.pathsep)]))
 
 
-EXTENSIONS = _get_path_extensions()
-_CONF_VAR_RE = re.compile(r"\{\w+}")
+EXTENSIONS: Final[list[str]] = _get_path_extensions()
+_CONF_VAR_RE: Final[re.Pattern[str]] = re.compile(
+    r"""
+    \{ \w+ \}   # sysconfig variable placeholder like {base}
+    """,
+    re.VERBOSE,
+)
 
 
 class PythonInfo:
@@ -594,7 +599,7 @@ def from_exe(
             except Exception as exception:
                 if raise_on_error:
                     raise
-                LOGGER.info("ignore %s due cannot resolve system due to %r", proposed.original_executable, exception)
+                _LOGGER.info("ignore %s due cannot resolve system due to %r", proposed.original_executable, exception)
                 proposed = None
         return proposed
 
@@ -619,12 +624,12 @@ def _resolve_to_system(cls, cache, target):
             prefix = target.real_prefix or target.base_prefix or target.prefix
             if prefix in prefixes:
                 if len(prefixes) == 1:
-                    LOGGER.info("%r links back to itself via prefixes", target)
+                    _LOGGER.info("%r links back to itself via prefixes", target)
                     target.system_executable = target.executable
                     break
                 for at, (p, t) in enumerate(prefixes.items(), start=1):
-                    LOGGER.error("%d: prefix=%s, info=%r", at, p, t)
-                LOGGER.error("%d: prefix=%s, info=%r", len(prefixes) + 1, prefix, target)
+                    _LOGGER.error("%d: prefix=%s, info=%r", at, p, t)
+                _LOGGER.error("%d: prefix=%s, info=%r", len(prefixes) + 1, prefix, target)
                 msg = "prefixes are causing a circle {}".format("|".join(prefixes.keys()))
                 raise RuntimeError(msg)
             prefixes[prefix] = target
@@ -657,9 +662,9 @@ def discover_exe(
         """
         key = prefix, exact
         if key in self._cache_exe_discovery and prefix:
-            LOGGER.debug("discover exe from cache %s - exact %s: %r", prefix, exact, self._cache_exe_discovery[key])
+            _LOGGER.debug("discover exe from cache %s - exact %s: %r", prefix, exact, self._cache_exe_discovery[key])
             return self._cache_exe_discovery[key]
-        LOGGER.debug("discover exe for %s in %s", self, prefix)
+        _LOGGER.debug("discover exe for %s in %s", self, prefix)
         possible_names = self._find_possible_exe_names()
         possible_folders = self._find_possible_folders(prefix)
         discovered = []
@@ -674,7 +679,7 @@ def discover_exe(
             info = self._select_most_likely(discovered, self)
             folders = os.pathsep.join(possible_folders)
             self._cache_exe_discovery[key] = info
-            LOGGER.debug("no exact match found, chosen most similar of %s within base folders %s", info, folders)
+            _LOGGER.debug("no exact match found, chosen most similar of %s within base folders %s", info, folders)
             return info
         msg = "failed to detect {} in {}".format("|".join(possible_names), os.pathsep.join(possible_folders))
         raise RuntimeError(msg)
@@ -693,7 +698,7 @@ def _check_exe(self, cache, folder, name, exact, discovered, env):
                 if item == "version_info":
                     found, searched = ".".join(str(i) for i in found), ".".join(str(i) for i in searched)
                 executable = info.executable
-                LOGGER.debug("refused interpreter %s because %s differs %s != %s", executable, item, found, searched)
+                _LOGGER.debug("refused interpreter %s because %s differs %s != %s", executable, item, found, searched)
                 if exact is False:
                     discovered.append(info)
                 break

src/py_discovery/_py_spec.py

@@ -106,7 +106,7 @@ def _int_or_none(val):
                     impl = specifier_match.group("impl")
                     spec_text = specifier_match.group("spec").strip()
                     try:
-                        version_specifier = SpecifierSet(spec_text)
+                        version_specifier = SpecifierSet.from_string(spec_text)
                     except InvalidSpecifier:  # pragma: no cover
                         pass
                     else:
@@ -171,7 +171,7 @@ def _check_version_specifier(self, spec: PythonSpec) -> bool:
         if spec.version_specifier is None:
             return True
         with contextlib.suppress(InvalidVersion):
-            Version(version_str)
+            Version.from_string(version_str)
             for item in spec.version_specifier:
                 required_precision = self._get_required_precision(item)
                 if required_precision is None or len(components) < required_precision: