✨ Venv recovery, requires-python reuse, --venv-dir, and logging

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(

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

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)

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"]