refactor(env_manager): split out python detection (#9050)

Author: finswimmer

src/poetry/console/commands/init.py

@@ -16,6 +16,7 @@
 from poetry.console.commands.command import Command
 from poetry.console.commands.env_command import EnvCommand
 from poetry.utils.dependency_specification import RequirementsParser
+from poetry.utils.env.python_manager import Python
 
 
 if TYPE_CHECKING:
@@ -96,7 +97,6 @@ def _init_pyproject(
         from poetry.config.config import Config
         from poetry.layouts import layout
         from poetry.pyproject.toml import PyProjectTOML
-        from poetry.utils.env import EnvManager
 
         is_interactive = self.io.is_interactive() and allow_interactive
 
@@ -174,11 +174,7 @@ def _init_pyproject(
             config = Config.create()
             python = (
                 ">="
-                + EnvManager.get_python_version(
-                    precision=2,
-                    prefer_active_python=config.get("virtualenvs.prefer-active-python"),
-                    io=self.io,
-                ).to_string()
+                + Python.get_preferred_python(config, self.io).minor_version.to_string()
             )
 
             if is_interactive:

src/poetry/utils/env/env_manager.py

@@ -5,7 +5,6 @@
 import os
 import plistlib
 import re
-import shutil
 import subprocess
 import sys
 
@@ -18,9 +17,7 @@
 import virtualenv
 
 from cleo.io.null_io import NullIO
-from cleo.io.outputs.output import Verbosity
 from poetry.core.constraints.version import Version
-from poetry.core.constraints.version import parse_constraint
 
 from poetry.toml.file import TOMLFile
 from poetry.utils._compat import WINDOWS
@@ -31,6 +28,7 @@
 from poetry.utils.env.exceptions import NoCompatiblePythonVersionFound
 from poetry.utils.env.exceptions import PythonVersionNotFound
 from poetry.utils.env.generic_env import GenericEnv
+from poetry.utils.env.python_manager import Python
 from poetry.utils.env.script_strings import GET_ENV_PATH_ONELINER
 from poetry.utils.env.script_strings import GET_PYTHON_VERSION_ONELINER
 from poetry.utils.env.system_env import SystemEnv
@@ -97,70 +95,6 @@ def __init__(self, poetry: Poetry, io: None | IO = None) -> None:
         self._poetry = poetry
         self._io = io or NullIO()
 
-    @staticmethod
-    def _full_python_path(python: str) -> Path | None:
-        # eg first find pythonXY.bat on windows.
-        path_python = shutil.which(python)
-        if path_python is None:
-            return None
-
-        try:
-            encoding = "locale" if sys.version_info >= (3, 10) else None
-            executable = subprocess.check_output(
-                [path_python, "-c", "import sys; print(sys.executable)"],
-                text=True,
-                encoding=encoding,
-            ).strip()
-            return Path(executable)
-
-        except CalledProcessError:
-            return None
-
-    @staticmethod
-    def _detect_active_python(io: None | IO = None) -> Path | None:
-        io = io or NullIO()
-        io.write_error_line(
-            "Trying to detect current active python executable as specified in"
-            " the config.",
-            verbosity=Verbosity.VERBOSE,
-        )
-
-        executable = EnvManager._full_python_path("python")
-
-        if executable is not None:
-            io.write_error_line(f"Found: {executable}", verbosity=Verbosity.VERBOSE)
-        else:
-            io.write_error_line(
-                "Unable to detect the current active python executable. Falling"
-                " back to default.",
-                verbosity=Verbosity.VERBOSE,
-            )
-
-        return executable
-
-    @staticmethod
-    def get_python_version(
-        precision: int = 3,
-        prefer_active_python: bool = False,
-        io: None | IO = None,
-    ) -> Version:
-        version = ".".join(str(v) for v in sys.version_info[:precision])
-
-        if prefer_active_python:
-            executable = EnvManager._detect_active_python(io)
-
-            if executable:
-                encoding = "locale" if sys.version_info >= (3, 10) else None
-                python_patch = subprocess.check_output(
-                    [executable, "-c", GET_PYTHON_VERSION_ONELINER],
-                    text=True,
-                    encoding=encoding,
-                ).strip()
-
-                version = ".".join(str(v) for v in python_patch.split(".")[:precision])
-
-        return Version.parse(version)
-
     @property
     def in_project_venv(self) -> Path:
         venv: Path = self._poetry.file.path.parent / ".venv"
@@ -189,24 +123,10 @@ def activate(self, python: str) -> Env:
             # Executable in PATH or full executable path
             pass
 
-        python_path = self._full_python_path(python)
-        if python_path is None:
+        python_ = Python.get_by_name(python)
+        if python_ is None:
             raise PythonVersionNotFound(python)
 
-        try:
-            encoding = "locale" if sys.version_info >= (3, 10) else None
-            python_version_string = subprocess.check_output(
-                [python_path, "-c", GET_PYTHON_VERSION_ONELINER],
-                text=True,
-                encoding=encoding,
-            )
-        except CalledProcessError as e:
-            raise EnvCommandError(e)
-
-        python_version = Version.parse(python_version_string.strip())
-        minor = f"{python_version.major}.{python_version.minor}"
-        patch = python_version.text
-
         create = False
         # If we are required to create the virtual environment in the project directory,
         # create or recreate it if needed
@@ -218,10 +138,10 @@ def activate(self, python: str) -> Env:
                 _venv = VirtualEnv(venv)
                 current_patch = ".".join(str(v) for v in _venv.version_info[:3])
 
-                if patch != current_patch:
+                if python_.patch_version.to_string() != current_patch:
                     create = True
 
-            self.create_venv(executable=python_path, force=create)
+            self.create_venv(executable=python_.executable, force=create)
 
             return self.get(reload=True)
 
@@ -233,11 +153,14 @@ def activate(self, python: str) -> Env:
                 current_minor = current_env["minor"]
                 current_patch = current_env["patch"]
 
-                if current_minor == minor and current_patch != patch:
+                if (
+                    current_minor == python_.minor_version.to_string()
+                    and current_patch != python_.patch_version.to_string()
+                ):
                     # We need to recreate
                     create = True
 
-        name = f"{self.base_env_name}-py{minor}"
+        name = f"{self.base_env_name}-py{python_.minor_version.to_string()}"
         venv = venv_path / name
 
         # Create if needed
@@ -251,13 +174,16 @@ def activate(self, python: str) -> Env:
                 _venv = VirtualEnv(venv)
                 current_patch = ".".join(str(v) for v in _venv.version_info[:3])
 
-                if patch != current_patch:
+                if python_.patch_version.to_string() != current_patch:
                     create = True
 
-            self.create_venv(executable=python_path, force=create)
+            self.create_venv(executable=python_.executable, force=create)
 
         # Activate
-        envs[self.base_env_name] = {"minor": minor, "patch": patch}
+        envs[self.base_env_name] = {
+            "minor": python_.minor_version.to_string(),
+            "patch": python_.patch_version.to_string(),
+        }
         self.envs_file.write(envs)
 
         return self.get(reload=True)
@@ -277,12 +203,8 @@ def get(self, reload: bool = False) -> Env:
         if self._env is not None and not reload:
             return self._env
 
-        prefer_active_python = self._poetry.config.get(
-            "virtualenvs.prefer-active-python"
-        )
-        python_minor = self.get_python_version(
-            precision=2, prefer_active_python=prefer_active_python, io=self._io
-        ).to_string()
+        python = Python.get_preferred_python(config=self._poetry.config, io=self._io)
+        python_minor = python.minor_version.to_string()
 
         env = None
         envs = None
@@ -480,8 +402,11 @@ def create_venv(
         )
         venv_prompt = self._poetry.config.get("virtualenvs.prompt")
 
-        if not executable and prefer_active_python:
-            executable = self._detect_active_python()
+        python = (
+            Python(executable)
+            if executable
+            else Python.get_preferred_python(config=self._poetry.config, io=self._io)
+        )
 
         venv_path = (
             self.in_project_venv
@@ -491,19 +416,8 @@ def create_venv(
         if not name:
             name = self._poetry.package.name
 
-        python_patch = ".".join([str(v) for v in sys.version_info[:3]])
-        python_minor = ".".join([str(v) for v in sys.version_info[:2]])
-        if executable:
-            encoding = "locale" if sys.version_info >= (3, 10) else None
-            python_patch = subprocess.check_output(
-                [executable, "-c", GET_PYTHON_VERSION_ONELINER],
-                text=True,
-                encoding=encoding,
-            ).strip()
-            python_minor = ".".join(python_patch.split(".")[:2])
-
         supported_python = self._poetry.package.python_constraint
-        if not supported_python.allows(Version.parse(python_patch)):
+        if not supported_python.allows(python.patch_version):
             # The currently activated or chosen Python version
             # is not compatible with the Python constraint specified
             # for the project.
@@ -512,71 +426,29 @@ def create_venv(
             # Otherwise, we try to find a compatible Python version.
             if executable and not prefer_active_python:
                 raise NoCompatiblePythonVersionFound(
-                    self._poetry.package.python_versions, python_patch
+                    self._poetry.package.python_versions,
+                    python.patch_version.to_string(),
                 )
 
             self._io.write_error_line(
-                f"<warning>The currently activated Python version {python_patch} is not"
+                f"<warning>The currently activated Python version {python.patch_version.to_string()} is not"
                 f" supported by the project ({self._poetry.package.python_versions}).\n"
                 "Trying to find and use a compatible version.</warning> "
             )
 
-            for suffix in sorted(
-                self._poetry.package.AVAILABLE_PYTHONS,
-                key=lambda v: (v.startswith("3"), -len(v), v),
-                reverse=True,
-            ):
-                if len(suffix) == 1:
-                    if not parse_constraint(f"^{suffix}.0").allows_any(
-                        supported_python
-                    ):
-                        continue
-                elif not supported_python.allows_any(parse_constraint(suffix + ".*")):
-                    continue
-
-                python_name = f"python{suffix}"
-                if self._io.is_debug():
-                    self._io.write_error_line(f"<debug>Trying {python_name}</debug>")
-
-                python = self._full_python_path(python_name)
-                if python is None:
-                    continue
-
-                try:
-                    encoding = "locale" if sys.version_info >= (3, 10) else None
-                    python_patch = subprocess.check_output(
-                        [python, "-c", GET_PYTHON_VERSION_ONELINER],
-                        stderr=subprocess.STDOUT,
-                        text=True,
-                        encoding=encoding,
-                    ).strip()
-                except CalledProcessError:
-                    continue
-
-                if supported_python.allows(Version.parse(python_patch)):
-                    self._io.write_error_line(
-                        f"Using <c1>{python_name}</c1> ({python_patch})"
-                    )
-                    executable = python
-                    python_minor = ".".join(python_patch.split(".")[:2])
-                    break
-
-            if not executable:
-                raise NoCompatiblePythonVersionFound(
-                    self._poetry.package.python_versions
-                )
+            python = Python.get_compatible_python(poetry=self._poetry, io=self._io)
 
         if in_project_venv:
             venv = venv_path
         else:
             name = self.generate_env_name(name, str(cwd))
-            name = f"{name}-py{python_minor.strip()}"
+            name = f"{name}-py{python.minor_version.to_string()}"
             venv = venv_path / name
 
         if venv_prompt is not None:
             venv_prompt = venv_prompt.format(
                 project_name=self._poetry.package.name or "virtualenv",
-                python_version=python_minor,
+                python_version=python.minor_version.to_string(),
             )
 
         if not venv.exists():
@@ -613,7 +485,7 @@ def create_venv(
         if create_venv:
             self.build_venv(
                 venv,
-                executable=executable,
+                executable=python.executable,
                 flags=self._poetry.config.get("virtualenvs.options"),
                 prompt=venv_prompt,
             )