Skip to content

Cache

cache

LRU cache for parsed PackageInfo objects.

Avoids redundant analyze_package calls when multiple tools query the same codebase within a single session.

The cache automatically invalidates entries when the set of .py files in the package directory changes — including content modifications (tracked via mtime).

Example::

>>> from axm_ast.core.cache import get_package, clear_cache
>>> pkg = get_package(Path("src/mylib"))  # parses on first call
>>> pkg2 = get_package(Path("src/mylib"))  # cache hit
>>> pkg is pkg2
True
>>> clear_cache()  # reset for re-parsing

PackageCache

Thread-safe cache for PackageInfo and call-site data.

Stores results keyed by resolved absolute path. On cache hit, the current .py file fingerprint (paths + mtime) is compared to the fingerprint recorded at parse time — if it differs the entry is evicted and the package is re-parsed.

Source code in packages/axm-ast/src/axm_ast/core/cache.py
class PackageCache:
    """Thread-safe cache for ``PackageInfo`` and call-site data.

    Stores results keyed by resolved absolute path.  On cache hit,
    the current ``.py`` file fingerprint (paths + mtime) is compared
    to the fingerprint recorded at parse time — if it differs the
    entry is evicted and the package is re-parsed.
    """

    def __init__(self) -> None:
        self._store: dict[Path, tuple[PackageInfo, _Fingerprint]] = {}
        self._calls_store: dict[Path, dict[str, list[CallSite]]] = {}
        self._lock = threading.Lock()

    def get(self, path: Path) -> PackageInfo:
        """Return cached ``PackageInfo`` or parse and cache on miss.

        On cache hit the file fingerprint is re-checked; if the set
        of ``.py`` files changed (addition, deletion, **or content
        modification**) the stale entry is evicted and the package
        is re-parsed.

        Args:
            path: Path to the package root directory.

        Returns:
            Cached or freshly parsed ``PackageInfo``.

        Raises:
            ValueError: If path is not a directory (from ``analyze_package``).
        """
        key = path.resolve()
        with self._lock:
            if key in self._store:
                cached_pkg, cached_fp = self._store[key]
                current_fp = _file_fingerprint(key)
                if current_fp == cached_fp:
                    return cached_pkg
                # Stale — evict package and call-sites together
                del self._store[key]
                self._calls_store.pop(key, None)

        # Parse outside the lock to avoid blocking other threads
        pkg = analyze_package(key)
        fp = _file_fingerprint(key)
        with self._lock:
            # Double-check: another thread may have populated it
            if key not in self._store:
                self._store[key] = (pkg, fp)
            return self._store[key][0]

    def get_calls(self, path: Path) -> dict[str, list[CallSite]]:
        """Return cached call-sites per module, extracting on first call.

        Call-sites share the same invalidation lifecycle as
        ``PackageInfo`` — when the fingerprint changes, both are
        evicted.

        Args:
            path: Path to the package root directory.

        Returns:
            Dict mapping dotted module names to their call-sites.
        """
        from axm_ast.core.callers import extract_calls

        key = path.resolve()
        # Ensure PackageInfo is cached (also handles fingerprint check)
        pkg = self.get(path)

        with self._lock:
            if key in self._calls_store:
                return self._calls_store[key]

        # Extract outside the lock
        calls_by_module: dict[str, list[CallSite]] = {}
        for mod in pkg.modules:
            mod_name = module_dotted_name(mod.path, pkg.root)
            calls_by_module[mod_name] = extract_calls(mod, module_name=mod_name)

        with self._lock:
            if key not in self._calls_store:
                self._calls_store[key] = calls_by_module
            return self._calls_store[key]

    def clear(self) -> None:
        """Invalidate all cached entries."""
        with self._lock:
            self._store.clear()
            self._calls_store.clear()
clear()

Invalidate all cached entries.

Source code in packages/axm-ast/src/axm_ast/core/cache.py
def clear(self) -> None:
    """Invalidate all cached entries."""
    with self._lock:
        self._store.clear()
        self._calls_store.clear()
get(path)

Return cached PackageInfo or parse and cache on miss.

On cache hit the file fingerprint is re-checked; if the set of .py files changed (addition, deletion, or content modification) the stale entry is evicted and the package is re-parsed.

Parameters:

Name Type Description Default
path Path

Path to the package root directory.

required

Returns:

Type Description
PackageInfo

Cached or freshly parsed PackageInfo.

Raises:

Type Description
ValueError

If path is not a directory (from analyze_package).

Source code in packages/axm-ast/src/axm_ast/core/cache.py
def get(self, path: Path) -> PackageInfo:
    """Return cached ``PackageInfo`` or parse and cache on miss.

    On cache hit the file fingerprint is re-checked; if the set
    of ``.py`` files changed (addition, deletion, **or content
    modification**) the stale entry is evicted and the package
    is re-parsed.

    Args:
        path: Path to the package root directory.

    Returns:
        Cached or freshly parsed ``PackageInfo``.

    Raises:
        ValueError: If path is not a directory (from ``analyze_package``).
    """
    key = path.resolve()
    with self._lock:
        if key in self._store:
            cached_pkg, cached_fp = self._store[key]
            current_fp = _file_fingerprint(key)
            if current_fp == cached_fp:
                return cached_pkg
            # Stale — evict package and call-sites together
            del self._store[key]
            self._calls_store.pop(key, None)

    # Parse outside the lock to avoid blocking other threads
    pkg = analyze_package(key)
    fp = _file_fingerprint(key)
    with self._lock:
        # Double-check: another thread may have populated it
        if key not in self._store:
            self._store[key] = (pkg, fp)
        return self._store[key][0]
get_calls(path)

Return cached call-sites per module, extracting on first call.

Call-sites share the same invalidation lifecycle as PackageInfo — when the fingerprint changes, both are evicted.

Parameters:

Name Type Description Default
path Path

Path to the package root directory.

required

Returns:

Type Description
dict[str, list[CallSite]]

Dict mapping dotted module names to their call-sites.

Source code in packages/axm-ast/src/axm_ast/core/cache.py
def get_calls(self, path: Path) -> dict[str, list[CallSite]]:
    """Return cached call-sites per module, extracting on first call.

    Call-sites share the same invalidation lifecycle as
    ``PackageInfo`` — when the fingerprint changes, both are
    evicted.

    Args:
        path: Path to the package root directory.

    Returns:
        Dict mapping dotted module names to their call-sites.
    """
    from axm_ast.core.callers import extract_calls

    key = path.resolve()
    # Ensure PackageInfo is cached (also handles fingerprint check)
    pkg = self.get(path)

    with self._lock:
        if key in self._calls_store:
            return self._calls_store[key]

    # Extract outside the lock
    calls_by_module: dict[str, list[CallSite]] = {}
    for mod in pkg.modules:
        mod_name = module_dotted_name(mod.path, pkg.root)
        calls_by_module[mod_name] = extract_calls(mod, module_name=mod_name)

    with self._lock:
        if key not in self._calls_store:
            self._calls_store[key] = calls_by_module
        return self._calls_store[key]

clear_cache()

Reset the global PackageCache, forcing re-parsing on next call.

Source code in packages/axm-ast/src/axm_ast/core/cache.py
def clear_cache() -> None:
    """Reset the global ``PackageCache``, forcing re-parsing on next call."""
    _cache.clear()

get_calls(path)

Return cached call-sites for path, using the global cache.

Equivalent to extracting calls from every module but avoids re-reading files when the package is already cached.

Parameters:

Name Type Description Default
path Path

Path to the package root directory.

required

Returns:

Type Description
dict[str, list[CallSite]]

Dict mapping dotted module names to call-site lists.

Source code in packages/axm-ast/src/axm_ast/core/cache.py
def get_calls(path: Path) -> dict[str, list[CallSite]]:
    """Return cached call-sites for *path*, using the global cache.

    Equivalent to extracting calls from every module but avoids
    re-reading files when the package is already cached.

    Args:
        path: Path to the package root directory.

    Returns:
        Dict mapping dotted module names to call-site lists.
    """
    return _cache.get_calls(path)

get_package(path)

Return PackageInfo for path, using the global cache.

Equivalent to analyze_package(path) but avoids re-parsing if the same path was already analyzed in this session.

Parameters:

Name Type Description Default
path Path

Path to the package root directory.

required

Returns:

Type Description
PackageInfo

Cached or freshly parsed PackageInfo.

Source code in packages/axm-ast/src/axm_ast/core/cache.py
def get_package(path: Path) -> PackageInfo:
    """Return ``PackageInfo`` for *path*, using the global cache.

    Equivalent to ``analyze_package(path)`` but avoids re-parsing
    if the same *path* was already analyzed in this session.

    Args:
        path: Path to the package root directory.

    Returns:
        Cached or freshly parsed ``PackageInfo``.
    """
    return _cache.get(path)