feat: Decouple FileCache from py_info (#2947)

Author: esafak

docs/changelog/2074a.feature.rst

@@ -0,0 +1 @@
+Decouple `FileCache` from `py_info` (discovery) - by :user:`esafak`.

src/virtualenv/cache/__init__.py

@@ -0,0 +1,9 @@
+from __future__ import annotations
+
+from .cache import Cache
+from .file_cache import FileCache
+
+__all__ = [
+    "Cache",
+    "FileCache",
+]

src/virtualenv/discovery/cache.py renamed to src/virtualenv/cache/cache.py

@@ -1,15 +1,17 @@
 from __future__ import annotations
 
 from abc import ABC, abstractmethod
-from typing import Any
+from typing import Any, Generic, Hashable, TypeVar
 
 try:
     from typing import Self  # pragma: ≥ 3.11 cover
 except ImportError:
     from typing_extensions import Self  # pragma: < 3.11 cover
 
+K = TypeVar("K", bound=Hashable)
 
-class Cache(ABC):
+
+class Cache(ABC, Generic[K]):
     """
     A generic cache interface.
 
@@ -18,7 +20,7 @@ class Cache(ABC):
     """
 
     @abstractmethod
-    def get(self, key: str) -> Any | None:
+    def get(self, key: K) -> Any | None:
         """
         Get a value from the cache.
 
@@ -28,7 +30,7 @@ def get(self, key: str) -> Any | None:
         raise NotImplementedError
 
     @abstractmethod
-    def set(self, key: str, value: Any) -> None:
+    def set(self, key: K, value: Any) -> None:
         """
         Set a value in the cache.
 
@@ -38,7 +40,7 @@ def set(self, key: str, value: Any) -> None:
         raise NotImplementedError
 
     @abstractmethod
-    def remove(self, key: str) -> None:
+    def remove(self, key: K) -> None:
         """
         Remove a value from the cache.
 

src/virtualenv/cache/file_cache.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from virtualenv.app_data.na import AppDataDisabled
+from virtualenv.cache import Cache
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+    from virtualenv.app_data.base import AppData
+
+
+class FileCache(Cache):
+    def __init__(self, app_data: AppData) -> None:
+        self.app_data = app_data if app_data is not None else AppDataDisabled()
+
+    def get(self, key: Path) -> dict | None:
+        """Get a value from the file cache."""
+        py_info, py_info_store = None, self.app_data.py_info(key)
+        with py_info_store.locked():
+            if py_info_store.exists():
+                py_info = py_info_store.read()
+        return py_info
+
+    def set(self, key: Path, value: dict) -> None:
+        """Set a value in the file cache."""
+        py_info_store = self.app_data.py_info(key)
+        with py_info_store.locked():
+            py_info_store.write(value)
+
+    def remove(self, key: Path) -> None:
+        """Remove a value from the file cache."""
+        py_info_store = self.app_data.py_info(key)
+        with py_info_store.locked():
+            if py_info_store.exists():
+                py_info_store.remove()
+
+    def clear(self) -> None:
+        """Clear the entire file cache."""
+        self.app_data.py_info_clear()
+
+
+__all__ = [
+    "FileCache",
+]

src/virtualenv/discovery/cached_py_info.py

@@ -7,6 +7,8 @@
 
 from __future__ import annotations
 
+import hashlib
+import importlib.util
 import logging
 import os
 import random
@@ -19,11 +21,11 @@
 from typing import TYPE_CHECKING
 
 from virtualenv.app_data.na import AppDataDisabled
-from virtualenv.discovery.file_cache import FileCache
+from virtualenv.cache import FileCache
 
 if TYPE_CHECKING:
     from virtualenv.app_data.base import AppData
-    from virtualenv.discovery.cache import Cache
+    from virtualenv.cache import Cache
 from virtualenv.discovery.py_info import PythonInfo
 from virtualenv.util.subprocess import subprocess
 
@@ -69,18 +71,42 @@ def _get_from_cache(cls, app_data: AppData, exe: str, env, cache: Cache, *, igno
 
 
 def _get_via_file_cache(cls, app_data: AppData, path: Path, exe: str, env, cache: Cache) -> PythonInfo:  # noqa: PLR0913
-    py_info = cache.get(path)
-    if py_info is not None:
-        py_info = cls._from_dict(py_info)
-        sys_exe = py_info.system_executable
-        if sys_exe is not None and not os.path.exists(sys_exe):
-            cache.remove(path)
-            py_info = None
+    # 1. get the hash of the probing script
+    spec = importlib.util.find_spec("virtualenv.discovery.py_info")
+    script = Path(spec.origin)
+    try:
+        py_info_hash = hashlib.sha256(script.read_bytes()).hexdigest()
+    except OSError:
+        py_info_hash = None
+
+    # 2. get the mtime of the python executable
+    try:
+        path_modified = path.stat().st_mtime
+    except OSError:
+        path_modified = -1
+
+    # 3. check if we have a valid cache entry
+    py_info = None
+    data = cache.get(path)
+    if data is not None:
+        if data.get("path") == str(path) and data.get("st_mtime") == path_modified and data.get("hash") == py_info_hash:
+            py_info = cls._from_dict(data.get("content"))
+            sys_exe = py_info.system_executable
+            if sys_exe is not None and not os.path.exists(sys_exe):
+                py_info = None  # if system executable is no longer there, this is not valid
+        if py_info is None:
+            cache.remove(path)  # if cache is invalid, remove it
 
     if py_info is None:  # if not loaded run and save
         failure, py_info = _run_subprocess(cls, exe, app_data, env)
         if failure is None:
-            cache.set(path, py_info._to_dict())  # noqa: SLF001
+            data = {
+                "st_mtime": path_modified,
+                "path": str(path),
+                "content": py_info._to_dict(),  # noqa: SLF001
+                "hash": py_info_hash,
+            }
+            cache.set(path, data)
         else:
             py_info = failure
     return py_info