Class IncrementalAstCache

src/pyspector/ast_cache.py:232–444

Three-level incremental AST cache. Parameters ---------- cache_dir : Path, optional Directory for the persistent (L2) disk cache. When *None* only the in-memory (L1) cache is active. If the directory cannot be created, a warning is issued and the cache ope

class IncrementalAstCache:
    """
    Three-level incremental AST cache.

    Parameters
    ----------
    cache_dir : Path, optional
        Directory for the persistent (L2) disk cache.  When *None* only the
        in-memory (L1) cache is active.  If the directory cannot be created,
        a warning is issued and the cache operates in L1-only mode.
    max_l1_entries : int
        Maximum entries kept in the in-memory LRU cache.  Oldest entries are
        evicted when the limit is exceeded.  Default: 512.

    Usage
    -----
    ::

        cache = IncrementalAstCache(cache_dir=Path(".pyspector_cache/ast"))
        ast_json = cache.get_ast_json(Path("src/foo.py"), content)
    """

    def __init__(
        self,
        cache_dir: Optional[Path] = None,
        max_l1_entries: int = MAX_L1_ENTRIES,
    ) -> None:
        self._l1: OrderedDict[str, FileCacheEntry] = OrderedDict()
        self._max_l1 = max_l1_entries
        self._cache_dir: Optional[Path] = None
        if cache_dir:
            try:
                cache_dir.mkdir(parents=True, exist_ok=True)
                self._cache_dir = cache_dir
            except OSError as e:
                warnings.warn(
                    f"PySpector: cannot create cache directory {cache_dir!r}: {e}. "
                    "Disk cache disabled for this run.",
                    stacklevel=2,
                )

    # ── Public API ───────────────────────────────────────────────────────────

    def get_ast_json(self, file_path: Path, content: str) -> str:
        """
        Return the AST JSON string for *file_path*.

        Raises
        ------
        SyntaxError
            If the file cannot be parsed, so callers can emit user-facing
            warnings while keeping cache logic out of the CLI layer.
        """
        return zlib.decompress(self._get_entry(file_path, content).full_ast_json_z).decode()

    def invalidate(self, file_path: Path) -> None:
        """Remove all cached data for a single file."""
        key = str(file_path.resolve())