diff --git a/idae/cli.py b/idae/cli.py index 65bc207..6b77d21 100644 --- a/idae/cli.py +++ b/idae/cli.py @@ -1,5 +1,6 @@ """CLI interface.""" import itertools +import logging import platform import shlex import subprocess @@ -12,7 +13,9 @@ from packaging.requirements import Requirement from packaging.version import Version from rich.console import Console +from rich.logging import RichHandler +from idae.dependencies import hash_dependencies from idae.pep723 import read from idae.resolver import get_python_or_exit from idae.venv import Python, clean_venvs, get_venv @@ -24,6 +27,22 @@ cli = typer.Typer() console = Console(stderr=True) +logger = logging.getLogger("idae") + + +def _setup_logging(verbose: int) -> None: + """Configure logging from a -v count (0=warning, 1=info, 2+=debug).""" + level = logging.WARNING + if verbose == 1: + level = logging.INFO + elif verbose >= 2: # noqa: PLR2004 + level = logging.DEBUG + logging.basicConfig( + level=level, + format="%(message)s", + datefmt="[%X]", + handlers=[RichHandler(console=console, show_path=False, rich_tracebacks=True)], + ) @cli.command(context_settings={"ignore_unknown_options": True}) @@ -71,11 +90,31 @@ def run( # noqa: PLR0913 help="Force idae to use a specific Python version", ), ] = None, + venv_dir: Annotated[ + Optional[Path], # noqa: FA100 + typer.Option( + "--venv-dir", + help="Create/reuse the venv at this directory instead of the cache", + file_okay=False, + dir_okay=True, + resolve_path=True, + ), + ] = None, + verbose: Annotated[ + int, + typer.Option( + "--verbose", + "-v", + count=True, + help="Increase verbosity (-v for info, -vv for debug + pip output)", + ), + ] = 0, ) -> None: """Automatically install necessary dependencies to run a Python script. --clean can be used without 'SCRIPT' """ + _setup_logging(verbose) if clean: clean_venvs() if script is None: @@ -92,23 +131,26 @@ def run( # noqa: PLR0913 ), executable=sys.executable, ) - if force_version is not None: - python = get_python_or_exit(force_version, console) if pyproject is not None: script_deps = ( [] if "dependencies" not in pyproject else list(map(Requirement, pyproject["dependencies"])) ) + dep_hash = hash_dependencies(script_deps) + if force_version is not None: + python = get_python_or_exit(force_version, console) + elif ( + not ignore_version and pyproject is not None and "requires-python" in pyproject + ): + # Prefer an existing cached venv whose Python satisfies the clause (#14) + python = get_python_or_exit( + pyproject["requires-python"], + console, + dep_hash=dep_hash, + ) - if ( - not ignore_version - and force_version is None - and "requires-python" in pyproject - ): - python = get_python_or_exit(pyproject["requires-python"], console) - - venv_path = get_venv(script_deps, python) + venv_path = get_venv(script_deps, python, venv_dir=venv_dir) extra_flags = list( itertools.chain.from_iterable( @@ -120,6 +162,7 @@ def run( # noqa: PLR0913 map(shlex.split, args or []), ), ) + logger.info("Running %s with %s", script, python.version) # Run the script inside the venv raise typer.Exit( code=subprocess.run( diff --git a/idae/resolver.py b/idae/resolver.py index da591ce..e160e12 100644 --- a/idae/resolver.py +++ b/idae/resolver.py @@ -2,20 +2,33 @@ from __future__ import annotations +import logging from typing import TYPE_CHECKING import findpython # type: ignore[import-untyped] import typer from packaging.specifiers import InvalidSpecifier, SpecifierSet +from .venv import CACHE_DIR, Python, cache_venv_path, is_venv_usable + if TYPE_CHECKING: # pragma: no cover from rich.console import Console +logger = logging.getLogger("idae") + + +def get_python_or_exit( + version: str, + console: Console, + dep_hash: str | None = None, +) -> findpython.PythonVersion: + """Return a PythonVersion or raise Exit. -def get_python_or_exit(version: str, console: Console) -> findpython.PythonVersion: - """Return a PythonVersion or raise Exit.""" + When ``dep_hash`` is given, an already-cached venv whose Python satisfies + ``version`` is preferred over creating a brand new one (issue #14). + """ try: - output = get_python(version) + output = get_python(version, dep_hash) except InvalidSpecifier as err: console.print(f"[red]error: Python version {version} could not be parsed[/red]") raise typer.Exit(code=1) from err @@ -25,18 +38,54 @@ def get_python_or_exit(version: str, console: Console) -> findpython.PythonVersi return output -def get_python(version: str) -> findpython.PythonVersion | None: - """Resolve the version string and return a valid Python.""" - # Order from latest version to earliest - pythons = {python.version: python for python in findpython.find_all()} +def _normalize_spec(version: str) -> SpecifierSet: try: float(version) except ValueError: pass else: version = f"~={float(version)}" - target = SpecifierSet(version) - for python_version, python in pythons.items(): - if python_version in target: + return SpecifierSet(version) + + +def _cached_python_for( + target: SpecifierSet, + dep_hash: str, + pythons: list[findpython.PythonVersion], +) -> findpython.PythonVersion | None: + """Find a Python that satisfies ``target`` and already has a cached venv.""" + if not CACHE_DIR.is_dir(): + return None + for python in pythons: + if python.version not in target: + continue + venv_path = cache_venv_path( + dep_hash, + Python(python.version, python.executable), + ) + if venv_path.is_dir() and is_venv_usable(venv_path): + logger.debug("Reusing cached %s venv for %s", python.version, target) + return python + return None + + +def get_python( + version: str, + dep_hash: str | None = None, +) -> findpython.PythonVersion | None: + """Resolve the version string and return a valid Python. + + If ``dep_hash`` is provided and an existing cached venv's Python satisfies + the clause, that Python is reused instead of picking the newest match. + """ + # Order from latest version to earliest + pythons = list(findpython.find_all()) + target = _normalize_spec(version) + if dep_hash is not None: + cached = _cached_python_for(target, dep_hash, pythons) + if cached is not None: + return cached + for python in pythons: + if python.version in target: return python return None diff --git a/idae/venv.py b/idae/venv.py index aea1858..05370f2 100644 --- a/idae/venv.py +++ b/idae/venv.py @@ -1,6 +1,7 @@ """Utils for venv creation.""" from __future__ import annotations +import logging import platform import shutil import subprocess @@ -20,6 +21,8 @@ CACHE_DIR = platformdirs.user_cache_path("idae") +logger = logging.getLogger("idae") + @dataclass class Python: @@ -29,45 +32,99 @@ class Python: executable: str | PathLike[str] -def get_venv(requirements: list[Requirement], python: Python) -> Path: - """Create or fetch a cached venv.""" - dep_hash = hash_dependencies(requirements) - venv_path = CACHE_DIR / f"{python.version.major}.{python.version.minor}" / dep_hash - if venv_path.is_dir(): - return venv_path - # This automatically includes pip - subprocess.run( - [python.executable, "-m", "venv", venv_path], # noqa: S603 - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, - check=True, - ) - # Install dependencies into the venv (if any) - if requirements: +def _bin_dir(venv_path: Path) -> Path: + return venv_path / ("Scripts" if platform.system() == "Windows" else "bin") + + +def is_venv_usable(venv_path: Path) -> bool: + """Return True if ``venv_path`` looks like a working virtual environment.""" + # A venv is broken if its interpreter is missing (issue #11). + return _bin_dir(venv_path).joinpath("python").exists() + + +def cache_venv_path(dep_hash: str, python: Python) -> Path: + """Return the cache location for a venv with the given deps and Python.""" + return CACHE_DIR / f"{python.version.major}.{python.version.minor}" / dep_hash + + +def _populate_venv( + venv_path: Path, + requirements: list[Requirement], + python: Python, +) -> None: + """Create a venv at ``venv_path`` and install ``requirements`` into it. + + Removes the partially-built venv if creation or installation fails or is + interrupted (issue #16). + """ + # Quiet by default; surface the subprocess output when verbose logging is on. + capture = logger.isEnabledFor(logging.DEBUG) + pipe = None if capture else subprocess.PIPE + try: + logger.debug("Creating venv at %s with %s", venv_path, python.executable) + # This automatically includes pip subprocess.run( - [ # noqa: S603 - ( - venv_path - / ("Scripts" if platform.system() == "Windows" else "bin") - / "pip" - ).resolve(), - "install", - *map(str, requirements), - ], - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT, + [python.executable, "-m", "venv", venv_path], # noqa: S603 + stdout=pipe, + stderr=subprocess.STDOUT if pipe else None, check=True, ) + # Install dependencies into the venv (if any) + if requirements: + logger.debug( + "Installing dependencies: %s", + ", ".join(map(str, requirements)), + ) + subprocess.run( + [ # noqa: S603 + (_bin_dir(venv_path) / "pip").resolve(), + "install", + *map(str, requirements), + ], + stdout=pipe, + stderr=subprocess.STDOUT if pipe else None, + check=True, + ) + except (subprocess.CalledProcessError, KeyboardInterrupt, OSError): + logger.warning( + "Setup failed or interrupted; removing broken venv %s", + venv_path, + ) + shutil.rmtree(venv_path, ignore_errors=True) + raise # The above works according to the Python docs: # > You don't specifically need to activate a virtual environment, # > as you can just specify the full path to that environment`s Python interpreter # > when invoking Python. Furthermore, all scripts installed in the environment # > should be runnable without activating it. # - https://docs.python.org/3/library/venv.html#how-venvs-work + + +def get_venv( + requirements: list[Requirement], + python: Python, + venv_dir: Path | None = None, +) -> Path: + """Create or fetch a cached venv. + + When ``venv_dir`` is given, the venv lives there instead of the global + cache (issue #15). + """ + dep_hash = hash_dependencies(requirements) + venv_path = venv_dir if venv_dir is not None else cache_venv_path(dep_hash, python) + if venv_path.is_dir(): + if is_venv_usable(venv_path): + logger.debug("Reusing existing venv at %s", venv_path) + return venv_path + # Broken leftover (e.g. missing bin/python); rebuild it (issue #11). + logger.warning("Found broken venv at %s; recreating", venv_path) + shutil.rmtree(venv_path, ignore_errors=True) + _populate_venv(venv_path, requirements, python) return venv_path def clean_venvs() -> None: """CLI command to delete the cache.""" + logger.debug("Cleaning venv cache at %s", CACHE_DIR) # Ignore errors like the directory not existing shutil.rmtree(CACHE_DIR, ignore_errors=True) diff --git a/pyproject.toml b/pyproject.toml index a3d6de8..c5abc99 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -99,5 +99,5 @@ ignore = [ ] [tool.ruff.lint.per-file-ignores] -"tests/**/*.py" = ["S101", "D", "ANN201"] +"tests/**/*.py" = ["S101", "S603", "D", "ANN201"] "docs/conf.py" = ["INP001", "A001"] diff --git a/tests/test_features.py b/tests/test_features.py new file mode 100644 index 0000000..d3ee7b8 --- /dev/null +++ b/tests/test_features.py @@ -0,0 +1,74 @@ +# ruff: noqa: ANN001, ANN202 +"""Tests for venv-dir, broken-venv handling, and verbosity (issues #10/#11/#15).""" +import platform +import shutil +import subprocess +import sys + +import platformdirs +import pytest +from packaging.version import Version +from typer.testing import CliRunner + +from idae.cli import cli +from idae.venv import Python, get_venv, is_venv_usable + +runner = CliRunner(mix_stderr=False) + +CACHE_DIR = platformdirs.user_cache_path("idae") + + +def _bin(venv_path): + return venv_path / ("Scripts" if platform.system() == "Windows" else "bin") + + +@pytest.fixture() +def empty_cache(): # noqa: PT004 + if CACHE_DIR.is_dir(): + shutil.rmtree(CACHE_DIR, ignore_errors=True) + + +@pytest.mark.usefixtures("empty_cache") +def test_venv_dir(capfd, tmp_path): + target = tmp_path / "myvenv" + result = runner.invoke( + cli, + ["--venv-dir", str(target), "tests/examples/echo.py", "hi"], + ) + out, _ = capfd.readouterr() + assert result.exit_code == 0 + assert out.replace("\r", "") == "hi\n" + # The venv was created at the requested location, not the cache. + assert is_venv_usable(target) + assert not CACHE_DIR.exists() + + +def test_broken_venv_is_recreated(tmp_path): + python = Python( + version=Version("3.12.0"), + executable=sys.executable, + ) + # Build a real venv, then break it by deleting the interpreter. + venv_path = get_venv([], python, venv_dir=tmp_path / "v") + assert is_venv_usable(venv_path) + for name in ("python", "python.exe"): + (_bin(venv_path) / name).unlink(missing_ok=True) + assert not is_venv_usable(venv_path) + # get_venv should notice the breakage and rebuild a usable venv. + venv_path = get_venv([], python, venv_dir=tmp_path / "v") + assert is_venv_usable(venv_path) + + +@pytest.mark.usefixtures("empty_cache") +def test_verbose_flag(): + # Rich's Console doesn't play well with CliRunner's stream capture, so run + # idae as a real subprocess to observe the -v logging on stderr. + proc = subprocess.run( + [sys.executable, "-m", "idae", "-v", "tests/examples/echo.py", "hello"], + capture_output=True, + text=True, + check=True, + ) + assert proc.stdout.replace("\r", "") == "hello\n" + # -v enables info logging, which reports the run on stderr. + assert "Running" in proc.stderr