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::

Text Only
>>> 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
Python
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
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._index_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, call-sites and index together
                del self._store[key]
                self._calls_store.pop(key, None)
                self._index_store.pop(key, None)

        # Capture the fingerprint BEFORE parsing so the stored fingerprint
        # reflects the file state the analysis actually saw. Capturing it after
        # ``analyze_package`` would bake a modification made *during* analysis
        # into the fingerprint, leaving the entry permanently stale (the next
        # ``get`` would see ``current_fp == cached_fp`` and never re-parse).
        fp = _file_fingerprint(key)
        # Parse outside the lock to avoid blocking other threads
        pkg = analyze_package(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.

        Concurrency: best-effort double-checked locking. The store is
        read under the lock, the call-sites are extracted *outside* the
        lock (to avoid blocking other threads), then re-checked under the
        lock before writing. Two concurrent first-callers may both extract,
        but the race is benign: the values are pure functions of the same
        ``PackageInfo``, so they are equivalent, and ``first write wins``
        (the re-check skips the second store). The cost is a wasted
        recompute, never corruption — so no lock is held during extraction.

        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 get_call_index(self, path: Path) -> dict[str, list[CallSite]]:
        """Return a cached ``symbol -> call-sites`` index for the package.

        Built once from :meth:`get_calls` and reused, turning a
        ``find_callers`` query into an O(1) dict lookup instead of an
        O(total-call-sites) linear scan. Shares the call-site invalidation
        lifecycle (evicted when the fingerprint changes).

        Insertion order matches ``get_calls`` iteration order (module order,
        then in-module call order), so each symbol's list is identical to the
        order a linear scan would have produced.

        Concurrency: best-effort double-checked locking, same pattern as
        :meth:`get_calls`. The index is built *outside* the lock and the
        store is re-checked under the lock before writing. Two concurrent
        first-callers may both build the index, but the race is benign: the
        index is a pure function of the same call-sites, so the results are
        equivalent, and ``first write wins`` (the re-check skips the second
        store). The cost is a wasted recompute, never corruption.

        Args:
            path: Path to the package root directory.

        Returns:
            Dict mapping symbol name to its call-sites.
        """
        key = path.resolve()
        calls_by_module = self.get_calls(path)

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

        index: dict[str, list[CallSite]] = {}
        for mod_calls in calls_by_module.values():
            for call in mod_calls:
                index.setdefault(call.symbol, []).append(call)

        with self._lock:
            if key not in self._index_store:
                self._index_store[key] = index
            return self._index_store[key]

    def clear(self) -> None:
        """Invalidate all cached entries."""
        from axm_ast.core.parser import clear_parse_cache

        with self._lock:
            self._store.clear()
            self._calls_store.clear()
            self._index_store.clear()
        clear_parse_cache()
clear()

Invalidate all cached entries.

Source code in packages/axm-ast/src/axm_ast/core/cache.py
Python
206
207
208
209
210
211
212
213
214
def clear(self) -> None:
    """Invalidate all cached entries."""
    from axm_ast.core.parser import clear_parse_cache

    with self._lock:
        self._store.clear()
        self._calls_store.clear()
        self._index_store.clear()
    clear_parse_cache()
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
Python
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
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, call-sites and index together
            del self._store[key]
            self._calls_store.pop(key, None)
            self._index_store.pop(key, None)

    # Capture the fingerprint BEFORE parsing so the stored fingerprint
    # reflects the file state the analysis actually saw. Capturing it after
    # ``analyze_package`` would bake a modification made *during* analysis
    # into the fingerprint, leaving the entry permanently stale (the next
    # ``get`` would see ``current_fp == cached_fp`` and never re-parse).
    fp = _file_fingerprint(key)
    # Parse outside the lock to avoid blocking other threads
    pkg = analyze_package(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_call_index(path)

Return a cached symbol -> call-sites index for the package.

Built once from :meth:get_calls and reused, turning a find_callers query into an O(1) dict lookup instead of an O(total-call-sites) linear scan. Shares the call-site invalidation lifecycle (evicted when the fingerprint changes).

Insertion order matches get_calls iteration order (module order, then in-module call order), so each symbol's list is identical to the order a linear scan would have produced.

Concurrency: best-effort double-checked locking, same pattern as :meth:get_calls. The index is built outside the lock and the store is re-checked under the lock before writing. Two concurrent first-callers may both build the index, but the race is benign: the index is a pure function of the same call-sites, so the results are equivalent, and first write wins (the re-check skips the second store). The cost is a wasted recompute, never corruption.

Parameters:

Name Type Description Default
path Path

Path to the package root directory.

 required

Returns:

Type Description
dict[str, list[CallSite]]

Dict mapping symbol name to its call-sites.
Source code in packages/axm-ast/src/axm_ast/core/cache.py
Python
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
def get_call_index(self, path: Path) -> dict[str, list[CallSite]]:
    """Return a cached ``symbol -> call-sites`` index for the package.

    Built once from :meth:`get_calls` and reused, turning a
    ``find_callers`` query into an O(1) dict lookup instead of an
    O(total-call-sites) linear scan. Shares the call-site invalidation
    lifecycle (evicted when the fingerprint changes).

    Insertion order matches ``get_calls`` iteration order (module order,
    then in-module call order), so each symbol's list is identical to the
    order a linear scan would have produced.

    Concurrency: best-effort double-checked locking, same pattern as
    :meth:`get_calls`. The index is built *outside* the lock and the
    store is re-checked under the lock before writing. Two concurrent
    first-callers may both build the index, but the race is benign: the
    index is a pure function of the same call-sites, so the results are
    equivalent, and ``first write wins`` (the re-check skips the second
    store). The cost is a wasted recompute, never corruption.

    Args:
        path: Path to the package root directory.

    Returns:
        Dict mapping symbol name to its call-sites.
    """
    key = path.resolve()
    calls_by_module = self.get_calls(path)

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

    index: dict[str, list[CallSite]] = {}
    for mod_calls in calls_by_module.values():
        for call in mod_calls:
            index.setdefault(call.symbol, []).append(call)

    with self._lock:
        if key not in self._index_store:
            self._index_store[key] = index
        return self._index_store[key]
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.

Concurrency: best-effort double-checked locking. The store is read under the lock, the call-sites are extracted outside the lock (to avoid blocking other threads), then re-checked under the lock before writing. Two concurrent first-callers may both extract, but the race is benign: the values are pure functions of the same PackageInfo, so they are equivalent, and first write wins (the re-check skips the second store). The cost is a wasted recompute, never corruption — so no lock is held during extraction.

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
Python
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
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.

    Concurrency: best-effort double-checked locking. The store is
    read under the lock, the call-sites are extracted *outside* the
    lock (to avoid blocking other threads), then re-checked under the
    lock before writing. Two concurrent first-callers may both extract,
    but the race is benign: the values are pure functions of the same
    ``PackageInfo``, so they are equivalent, and ``first write wins``
    (the re-check skips the second store). The cost is a wasted
    recompute, never corruption — so no lock is held during extraction.

    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
Python
264
265
266
def clear_cache() -> None:
    """Reset the global ``PackageCache``, forcing re-parsing on next call."""
    _cache.clear()

get_call_index(path)

Return the cached symbol -> call-sites index, using the global cache.

Parameters:

Name Type Description Default
path Path

Path to the package root directory.

 required

Returns:

Type Description
dict[str, list[CallSite]]

Dict mapping symbol name to its call-sites.
Source code in packages/axm-ast/src/axm_ast/core/cache.py
Python
252
253
254
255
256
257
258
259
260
261
def get_call_index(path: Path) -> dict[str, list[CallSite]]:
    """Return the cached ``symbol -> call-sites`` index, using the global cache.

    Args:
        path: Path to the package root directory.

    Returns:
        Dict mapping symbol name to its call-sites.
    """
    return _cache.get_call_index(path)

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
Python
237
238
239
240
241
242
243
244
245
246
247
248
249
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
Python
222
223
224
225
226
227
228
229
230
231
232
233
234
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)