From 871d3796299af58f53b1dc47aea011438dfc23a3 Mon Sep 17 00:00:00 2001 From: Jack Cherng Date: Sun, 11 Aug 2024 11:57:12 +0800 Subject: [PATCH] fix: Conda venv is not used Conda doesn't follow PEP 405. Signed-off-by: Jack Cherng --- plugin/client.py | 11 +- plugin/venv_finder.py | 348 ++++++++++++++++++++++++------------------ 2 files changed, 210 insertions(+), 149 deletions(-) diff --git a/plugin/client.py b/plugin/client.py index 71c36b1..3eb282d 100644 --- a/plugin/client.py +++ b/plugin/client.py @@ -21,14 +21,19 @@ from .constants import PACKAGE_NAME from .log import log_info, log_warning from .template import load_string_template -from .venv_finder import VenvInfo, find_venv_by_finder_names, get_finder_name_mapping +from .venv_finder import ( + BaseVenvInfo, + find_venv_by_finder_names, + find_venv_by_python_executable, + get_finder_name_mapping, +) @dataclass class WindowAttr: simple_python_executable: Path | None = None """The path to the Python executable found by the `PATH` env variable.""" - venv_info: VenvInfo | None = None + venv_info: BaseVenvInfo | None = None """The information of the virtual environment.""" @property @@ -272,7 +277,7 @@ def _update_venv_info() -> None: window_attr.venv_info = None if python_path := settings.get("python.pythonPath"): - window_attr.venv_info = VenvInfo.from_python_executable(python_path) + window_attr.venv_info = find_venv_by_python_executable(python_path) return supported_finder_names = tuple(get_finder_name_mapping().keys()) diff --git a/plugin/venv_finder.py b/plugin/venv_finder.py index a61fdf7..5a7d8ed 100644 --- a/plugin/venv_finder.py +++ b/plugin/venv_finder.py @@ -1,15 +1,17 @@ from __future__ import annotations import configparser +import json import os import shutil import subprocess from abc import ABC, abstractmethod from dataclasses import dataclass, field from functools import lru_cache +from itertools import product from pathlib import Path from types import MappingProxyType -from typing import Any, Generator, Iterable, Mapping, Sequence, final +from typing import Any, Generator, Mapping, Sequence, final from more_itertools import first_true from typing_extensions import Self @@ -18,7 +20,8 @@ from .utils import camel_to_snake, get_default_startupinfo, iterate_by_line, remove_suffix -def find_venv_by_finder_names(finder_names: Sequence[str], *, project_dir: Path) -> VenvInfo | None: +def find_venv_by_finder_names(finder_names: Sequence[str], *, project_dir: Path) -> BaseVenvInfo | None: + """Finds the virtual environment information by finders.""" if isinstance(finder_names, str): finder_names = (finder_names,) @@ -32,6 +35,16 @@ def find_venv_by_finder_names(finder_names: Sequence[str], *, project_dir: Path) return None +def find_venv_by_python_executable(python_executable: str | Path) -> BaseVenvInfo | None: + """Finds the virtual environment information by the Python executable path.""" + return first_true( + map( + lambda cls: cls.from_python_executable(python_executable), + list_venv_info_classes(), + ), + ) + + @lru_cache def find_finder_class_by_name(name: str) -> type[BaseVenvFinder] | None: """Finds the virtual environment finder class by its name.""" @@ -44,8 +57,22 @@ def get_finder_name_mapping() -> Mapping[str, type[BaseVenvFinder]]: return MappingProxyType({finder_cls.name(): finder_cls for finder_cls in list_venv_finder_classes()}) +def list_venv_info_classes() -> Generator[type[BaseVenvInfo], None, None]: + """ + Lists all virtual environment information classes. + + The order matters because they will be used for testing one by one. + """ + yield Pep405VenvInfo + yield CondaVenvInfo + + def list_venv_finder_classes() -> Generator[type[BaseVenvFinder], None, None]: - """Lists all virtual environment finder classes. The order matters.""" + """ + Lists all virtual environment finder classes. + + The order matters because they will be used for testing one by one. + """ yield LocalDotVenvVenvFinder yield EnvVarCondaPrefixVenvFinder yield EnvVarVirtualEnvVenvFinder @@ -58,10 +85,26 @@ def list_venv_finder_classes() -> Generator[type[BaseVenvFinder], None, None]: yield AnySubdirectoryVenvFinder -@dataclass -class VenvInfoCache: - pyvenv_cfg: dict[str, Any] = field(default_factory=dict) - """The parsed results of the `pyvenv.cfg` file.""" +def run_shell_command(command: str, *, cwd: Path | None = None) -> tuple[str, str, int] | None: + try: + proc = subprocess.Popen( + command, + cwd=cwd, + shell=True, + startupinfo=get_default_startupinfo(), + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + universal_newlines=True, + ) + stdout, stderr = map(str.rstrip, proc.communicate()) + except Exception as e: + log_error(f"Failed running command ({command}): {e}") + return None + + if stderr: + log_error(f"Failed running command ({command}): {stderr}") + + return stdout, stderr, proc.returncode or 0 @dataclass @@ -70,24 +113,20 @@ class VenvInfoMeta: """The name of the virtual environment finder.""" -@dataclass(frozen=True) -class VenvInfo: +@dataclass +class BaseVenvInfo(ABC): """The information of the virtual environment.""" venv_dir: Path """The path of the virtual environment directory.""" - cache: VenvInfoCache = field(default_factory=VenvInfoCache) - """The cache.""" - meta: VenvInfoMeta = field(default_factory=VenvInfoMeta) - """The metadata.""" + prompt: str = "" + """The prompt of the virtual environment.""" + python_version: str = "" + """The Python version of the virtual environment.""" - @property - def prompt(self) -> str: - """The prompt of the virtual environment.""" - if prompt := str(self.cache.pyvenv_cfg.get("prompt", "")): - return prompt - return self.venv_dir.name + meta: VenvInfoMeta = field(default_factory=VenvInfoMeta) + """The metadata which is not related to venv.""" @property def python_executable(self) -> Path: @@ -96,52 +135,93 @@ def python_executable(self) -> Path: return self.venv_dir / "Scripts/python.exe" return self.venv_dir / "bin/python" - @property - def python_version(self) -> str: - """The Python version of the virtual environment.""" - # "venv" module uses "version" - if version := str(self.cache.pyvenv_cfg.get("version", "")): - return version - # "uv" utility uses "version_info" - if version := str(self.cache.pyvenv_cfg.get("version_info", "")): - return version - return "" - - @property - def pyvenv_cfg_path(self) -> Path: - """The path of the `pyvenv.cfg` file of the virtual environment.""" - return self.venv_dir / "pyvenv.cfg" - + @abstractmethod def is_valid(self) -> bool: """Checks if this virtual environment is valid.""" - try: - return self.venv_dir.is_dir() and self.pyvenv_cfg_path.is_file() and self.python_executable.is_file() - except PermissionError: - return False - def refresh_cache(self) -> None: - """Refreshes cached property values.""" - self.cache.pyvenv_cfg = self.parse_pyvenv_cfg(self.pyvenv_cfg_path) + @abstractmethod + def refresh_derived_attributes(self) -> None: + """Refreshes the derived attributes.""" + @classmethod + def from_python_executable(cls, python_executable: str | Path) -> Self | None: + """Create an instance from the Python executable path.""" + try: + venv_dir = Path(python_executable).parents[1] + except IndexError: + return None + return cls.from_venv_dir(venv_dir) + + @final @classmethod def from_venv_dir(cls, venv_dir: str | Path) -> Self | None: + """Create an instance from the virtual environment directory.""" try: venv_dir = Path(venv_dir).expanduser().resolve() except PermissionError: return None - if (venv_info := cls(venv_dir=venv_dir)).is_valid(): - venv_info.refresh_cache() - return venv_info - return None + if not (venv_info := cls(venv_dir=venv_dir)).is_valid(): + return None - @classmethod - def from_python_executable(cls, python_executable: str | Path) -> Self | None: + venv_info.refresh_derived_attributes() + return venv_info + + +class CondaVenvInfo(BaseVenvInfo): + """Venv information for Conda virtual environment.""" + + @property + def conda_meta_path(self) -> Path: + """The path of the `conda-meta` directory of the virtual environment.""" + return self.venv_dir / "conda-meta" + + def is_valid(self) -> bool: try: - venv_dir = Path(python_executable).parents[1] - except IndexError: + return self.python_executable.is_file() and self.conda_meta_path.is_dir() + except PermissionError: + return False + + def refresh_derived_attributes(self) -> None: + if not (output := run_shell_command("conda info --json")): return None - return cls.from_venv_dir(venv_dir) + stdout, _, _ = output + + try: + conda_info: dict[str, Any] = json.loads(stdout) + except json.JSONDecodeError: + return None + + self.prompt = conda_info.get("active_prefix_name", "") + self.python_version = ( + conda_info.get("python_version", "") + .replace(".alpha.", "a") + .replace(".beta.", "b") + .replace(".candidate.", "rc") + .partition(".final.")[0] + ) + + +class Pep405VenvInfo(BaseVenvInfo): + """Venv information for PEP 405 (https://peps.python.org/pep-0405/)""" + + @property + def pyvenv_cfg_path(self) -> Path: + """The path of the `pyvenv.cfg` file of the virtual environment.""" + return self.venv_dir / "pyvenv.cfg" + + def is_valid(self) -> bool: + try: + return self.python_executable.is_file() and self.pyvenv_cfg_path.is_file() + except PermissionError: + return False + + def refresh_derived_attributes(self) -> None: + pyvenv_cfg = self.parse_pyvenv_cfg(self.pyvenv_cfg_path) + + self.prompt = pyvenv_cfg.get("prompt", "") or self.venv_dir.name + # "venv" module uses "version" and "uv" utility uses "version_info" + self.python_version = pyvenv_cfg.get("version", "") or pyvenv_cfg.get("version_info", "") @classmethod def from_pyvenv_cfg_file(cls, pyvenv_cfg_file: str | Path) -> Self | None: @@ -173,81 +253,40 @@ def __init__(self, project_dir: Path) -> None: def name(cls) -> str: return camel_to_snake(remove_suffix(cls.__name__, "VenvFinder")) - @final @classmethod + @abstractmethod def can_support(cls, project_dir: Path) -> bool: """Check if this class support the given `project_dir`.""" - try: - return cls._can_support(project_dir) - except PermissionError: - return False @final - def find_venv(self) -> VenvInfo | None: + def find_venv(self) -> BaseVenvInfo | None: """Find the virtual environment.""" try: - venv_info = self._find_venv() + if not (venv_info := self.find_venv_()): + return None except PermissionError: return None - if venv_info: - venv_info.meta.finder_name = self.name() - return venv_info - return None - - @classmethod - @abstractmethod - def _can_support(cls, project_dir: Path) -> bool: - """Check if this class support the given `project_dir`. Implement this method by the subclass.""" + venv_info.meta.finder_name = self.name() + return venv_info @abstractmethod - def _find_venv(self) -> VenvInfo | None: + def find_venv_(self) -> BaseVenvInfo | None: """Find the virtual environment. Implement this method by the subclass.""" - @staticmethod - def _find_from_venv_dirs(venv_dirs: Iterable[Path]) -> VenvInfo | None: - def _filtered_candidates() -> Generator[Path, None, None]: - for venv_dir in venv_dirs: - try: - if venv_dir.is_dir(): - yield venv_dir - except PermissionError: - pass - - return first_true(map(VenvInfo.from_venv_dir, _filtered_candidates())) - - @staticmethod - def _run_shell_command(command: str, *, cwd: Path | None = None) -> tuple[str, str, int] | None: - try: - proc = subprocess.Popen( - command, - cwd=cwd, - shell=True, - startupinfo=get_default_startupinfo(), - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - universal_newlines=True, - ) - stdout, stderr = map(str.rstrip, proc.communicate()) - except Exception as e: - log_error(f"Failed running command ({command}): {e}") - return None - - if stderr: - log_error(f"Failed running command ({command}): {stderr}") - - return stdout, stderr, proc.returncode - class AnySubdirectoryVenvFinder(BaseVenvFinder): """Finds the virtual environment with any subdirectory.""" @classmethod - def _can_support(cls, project_dir: Path) -> bool: + def can_support(cls, project_dir: Path) -> bool: return True - def _find_venv(self) -> VenvInfo | None: - return self._find_from_venv_dirs(self.project_dir.iterdir()) + def find_venv_(self) -> BaseVenvInfo | None: + for subproject_dir, venv_info_cls in product(self.project_dir.iterdir(), list_venv_info_classes()): + if venv_info := venv_info_cls.from_venv_dir(subproject_dir): + return venv_info + return None class EnvVarCondaPrefixVenvFinder(BaseVenvFinder): @@ -258,11 +297,11 @@ class EnvVarCondaPrefixVenvFinder(BaseVenvFinder): """ @classmethod - def _can_support(cls, project_dir: Path) -> bool: + def can_support(cls, project_dir: Path) -> bool: return "CONDA_PREFIX" in os.environ - def _find_venv(self) -> VenvInfo | None: - return VenvInfo.from_venv_dir(os.environ["CONDA_PREFIX"]) + def find_venv_(self) -> CondaVenvInfo | None: + return CondaVenvInfo.from_venv_dir(os.environ["CONDA_PREFIX"]) class EnvVarVirtualEnvVenvFinder(BaseVenvFinder): @@ -273,11 +312,11 @@ class EnvVarVirtualEnvVenvFinder(BaseVenvFinder): """ @classmethod - def _can_support(cls, project_dir: Path) -> bool: + def can_support(cls, project_dir: Path) -> bool: return "VIRTUAL_ENV" in os.environ - def _find_venv(self) -> VenvInfo | None: - return VenvInfo.from_venv_dir(os.environ["VIRTUAL_ENV"]) + def find_venv_(self) -> Pep405VenvInfo | None: + return Pep405VenvInfo.from_venv_dir(os.environ["VIRTUAL_ENV"]) class LocalDotVenvVenvFinder(BaseVenvFinder): @@ -288,14 +327,16 @@ class LocalDotVenvVenvFinder(BaseVenvFinder): """ @classmethod - def _can_support(cls, project_dir: Path) -> bool: + def can_support(cls, project_dir: Path) -> bool: return True - def _find_venv(self) -> VenvInfo | None: - return self._find_from_venv_dirs(( - self.project_dir / ".venv", - self.project_dir / "venv", - )) + def find_venv_(self) -> Pep405VenvInfo | None: + return first_true( + map( + Pep405VenvInfo.from_venv_dir, + (self.project_dir / ".venv", self.project_dir / "venv"), + ) + ) class HatchVenvFinder(BaseVenvFinder): @@ -306,12 +347,12 @@ class HatchVenvFinder(BaseVenvFinder): """ @classmethod - def _can_support(cls, project_dir: Path) -> bool: + def can_support(cls, project_dir: Path) -> bool: return bool(shutil.which("hatch")) - def _find_venv(self) -> VenvInfo | None: + def find_venv_(self) -> Pep405VenvInfo | None: # "hatch env find" will always provide a calculated path, where the hatch-managed venv should be at - if not (output := self._run_shell_command("hatch env find", cwd=self.project_dir)): + if not (output := run_shell_command("hatch env find", cwd=self.project_dir)): return None venv_dir, _, exit_code = output @@ -319,7 +360,7 @@ def _find_venv(self) -> VenvInfo | None: # E.g., you run "hatch env find" in root `/` directory. if exit_code != 0 or not venv_dir: return None - return VenvInfo.from_venv_dir(venv_dir) + return Pep405VenvInfo.from_venv_dir(venv_dir) class PdmVenvFinder(BaseVenvFinder): @@ -330,17 +371,20 @@ class PdmVenvFinder(BaseVenvFinder): """ @classmethod - def _can_support(cls, project_dir: Path) -> bool: - return bool(shutil.which("pdm") and (project_dir / ".pdm-python").is_file()) + def can_support(cls, project_dir: Path) -> bool: + try: + return bool(shutil.which("pdm") and (project_dir / ".pdm-python").is_file()) + except Exception: + return False - def _find_venv(self) -> VenvInfo | None: - if not (output := self._run_shell_command("pdm info --python", cwd=self.project_dir)): + def find_venv_(self) -> Pep405VenvInfo | None: + if not (output := run_shell_command("pdm info --python", cwd=self.project_dir)): return None python_executable, _, _ = output if not python_executable: return None - return VenvInfo.from_python_executable(python_executable) + return Pep405VenvInfo.from_python_executable(python_executable) class PipenvVenvFinder(BaseVenvFinder): @@ -351,34 +395,40 @@ class PipenvVenvFinder(BaseVenvFinder): """ @classmethod - def _can_support(cls, project_dir: Path) -> bool: - return bool(shutil.which("pipenv") and (project_dir / "Pipfile").is_file()) + def can_support(cls, project_dir: Path) -> bool: + try: + return bool(shutil.which("pipenv") and (project_dir / "Pipfile").is_file()) + except Exception: + return False - def _find_venv(self) -> VenvInfo | None: - if not (output := self._run_shell_command("pipenv --py", cwd=self.project_dir)): + def find_venv_(self) -> Pep405VenvInfo | None: + if not (output := run_shell_command("pipenv --py", cwd=self.project_dir)): return None python_executable, _, _ = output if not python_executable: return None - return VenvInfo.from_python_executable(python_executable) + return Pep405VenvInfo.from_python_executable(python_executable) class PoetryVenvFinder(BaseVenvFinder): """Finds the virtual environment using `poetry`.""" @classmethod - def _can_support(cls, project_dir: Path) -> bool: - return bool(shutil.which("poetry") and (project_dir / "poetry.lock").is_file()) + def can_support(cls, project_dir: Path) -> bool: + try: + return bool(shutil.which("poetry") and (project_dir / "poetry.lock").is_file()) + except Exception: + return False - def _find_venv(self) -> VenvInfo | None: - if not (output := self._run_shell_command("poetry env info -p", cwd=self.project_dir)): + def find_venv_(self) -> Pep405VenvInfo | None: + if not (output := run_shell_command("poetry env info -p", cwd=self.project_dir)): return None venv_dir, _, _ = output if not venv_dir: return None - return VenvInfo.from_venv_dir(venv_dir) + return Pep405VenvInfo.from_venv_dir(venv_dir) class PyenvVenvFinder(BaseVenvFinder): @@ -389,17 +439,20 @@ class PyenvVenvFinder(BaseVenvFinder): """ @classmethod - def _can_support(cls, project_dir: Path) -> bool: - return bool(shutil.which("pyenv") and (project_dir / ".python-version").is_file()) + def can_support(cls, project_dir: Path) -> bool: + try: + return bool(shutil.which("pyenv") and (project_dir / ".python-version").is_file()) + except Exception: + return False - def _find_venv(self) -> VenvInfo | None: - if not (output := self._run_shell_command("pyenv which python", cwd=self.project_dir)): + def find_venv_(self) -> Pep405VenvInfo | None: + if not (output := run_shell_command("pyenv which python", cwd=self.project_dir)): return None python_executable, _, _ = output if not python_executable: return None - return VenvInfo.from_python_executable(python_executable) + return Pep405VenvInfo.from_python_executable(python_executable) class RyeVenvFinder(BaseVenvFinder): @@ -410,16 +463,19 @@ class RyeVenvFinder(BaseVenvFinder): """ @classmethod - def _can_support(cls, project_dir: Path) -> bool: - return bool(shutil.which("rye") and (project_dir / "pyproject.toml").is_file()) + def can_support(cls, project_dir: Path) -> bool: + try: + return bool(shutil.which("rye") and (project_dir / "pyproject.toml").is_file()) + except Exception: + return False - def _find_venv(self) -> VenvInfo | None: - if not (output := self._run_shell_command("rye show", cwd=self.project_dir)): + def find_venv_(self) -> Pep405VenvInfo | None: + if not (output := run_shell_command("rye show", cwd=self.project_dir)): return None stdout, _, _ = output for line in iterate_by_line(stdout): pre, sep, post = line.partition(":") if sep and pre == "venv": - return VenvInfo.from_venv_dir(post.strip()) + return Pep405VenvInfo.from_venv_dir(post.strip()) return None