diff --git a/src/openultrasast/hunt/producers.py b/src/openultrasast/hunt/producers.py index f4878f9..4053c2d 100644 --- a/src/openultrasast/hunt/producers.py +++ b/src/openultrasast/hunt/producers.py @@ -56,6 +56,7 @@ class Context: spans: dict[str, tuple[source.FunctionRange, ...]] = field(default_factory=dict) range_errors: dict[str, str] = field(default_factory=dict) irs: dict[str, FileIR] = field(default_factory=dict) + call_index: dict[str, tuple[source.CodeLine, ...]] = field(default_factory=dict) diff_bytes: int = 0 @cached_property @@ -211,7 +212,9 @@ def _source(root: Path, change: Change, path: str, line: int, ctx: Context) -> d ctx.read(path) if path in ctx.range_errors: raise _Absent(ctx.range_errors[path]) - facts = source.facts_for(root, path, line, entry_points=ctx.entry_points, files=ctx.product_files, texts=ctx.texts) + facts = source.facts_for( + root, path, line, entry_points=ctx.entry_points, files=ctx.product_files, texts=ctx.texts, index=ctx.call_index + ) # Select the occurrence at this line when several methods share a name. spans = [r for r in ctx.ranges(path) if r.name == facts.function and r.start <= line <= r.end] span = min(spans, key=lambda r: r.end - r.start) if spans else None diff --git a/src/openultrasast/source.py b/src/openultrasast/source.py index 6149080..c830d0b 100644 --- a/src/openultrasast/source.py +++ b/src/openultrasast/source.py @@ -4,7 +4,7 @@ from __future__ import annotations import hashlib import re -from collections.abc import Iterable, Mapping, Sequence +from collections.abc import Iterable, Mapping, MutableMapping, Sequence from pathlib import Path from types import MappingProxyType from typing import TYPE_CHECKING, Any, NamedTuple @@ -389,19 +389,37 @@ def spans_named(ranges: Sequence[FunctionRange], name: str) -> tuple[tuple[int, return tuple((item.start, item.end) for item in ranges if item.name == name) -def call_sites(text: str, names: Iterable[str]) -> tuple[CallSite, ...]: - """Return calls in line order, excluding declarations and comments, once per name and line.""" - wanted = sorted(set(names) - RESERVED_FUNCTION_LABELS - {MODULE_FUNCTION}, key=lambda name: (-len(name), name)) - if not wanted: - return () - token = re.compile(r"(? tuple[CodeLine, ...]: + """Prepare a file for repeated call searches; call_sites over many names of one file pays the declaration scan once.""" + out = [] for number, line in enumerate(_code_lines(text), start=1): declared = { next((group for group in match.groups() if group), "") for pattern in DECLARATION.values() if (match := pattern.match(line)) } if "{" in line or "=>" in line: declared.add(declarator_name(line)) + out.append(CodeLine(number, line, frozenset(name for name in declared if name))) + return tuple(out) + + +def call_sites(text: str, names: Iterable[str], *, index: Sequence[CodeLine] | None = None) -> tuple[CallSite, ...]: + """Return calls in line order, excluding declarations and comments, once per name and line. + + ``index`` is ``code_index(text)`` prepared by the caller; without it the file is scanned here.""" + wanted = sorted(set(names) - RESERVED_FUNCTION_LABELS - {MODULE_FUNCTION}, key=lambda name: (-len(name), name)) + if not wanted: + return () + token = re.compile(r"(? str: return resolved.read_text(encoding="utf-8", errors="replace") -def callers_of(root: Path, name: str, *, files: Iterable[str], texts: Mapping[str, str] | None = None) -> tuple[tuple[str, int], ...]: +def callers_of( + root: Path, + name: str, + *, + files: Iterable[str], + texts: Mapping[str, str] | None = None, + index: MutableMapping[str, tuple[CodeLine, ...]] | None = None, +) -> tuple[tuple[str, int], ...]: """Return sorted call locations from the supplied repository-relative product files. - ``texts`` holds file contents already read by the caller; files absent from it are read here.""" + ``texts`` holds file contents already read by the caller; files absent from it are read here. ``index`` is a + cache of prepared files the caller keeps across calls (one checkout, many names); it is filled on first use.""" root = Path(root) def text_of(path: str) -> str: return texts[path] if texts is not None and path in texts else _read_source(root, path) - return tuple((path, call.line) for path in sorted(set(files)) for call in call_sites(text_of(path), (name,))) + def index_of(path: str) -> tuple[CodeLine, ...]: + if index is None: + return code_index(text_of(path)) + if path not in index: + index[path] = code_index(text_of(path)) + return index[path] + + return tuple((path, call.line) for path in sorted(set(files)) for call in call_sites("", (name,), index=index_of(path))) def facts_for( @@ -440,6 +473,7 @@ def facts_for( entry_points: Sequence[EntryPointRecord] | None = None, files: Sequence[str] | None = None, texts: Mapping[str, str] | None = None, + index: MutableMapping[str, tuple[CodeLine, ...]] | None = None, ) -> SourceFacts: """Compute repository facts at a line; supplied entry points, product files and texts avoid repeated work. @@ -456,7 +490,7 @@ def facts_for( entry_points = analyze_entry_points(root, [build_file_target(root, root / file) for file in files]) return SourceFacts( - language, function, bounds(spans, function), callers_of(root, function, files=files, texts=texts), tuple(entry_points) + language, function, bounds(spans, function), callers_of(root, function, files=files, texts=texts, index=index), tuple(entry_points) ) diff --git a/tests/test_source.py b/tests/test_source.py index 9545a16..374c768 100644 --- a/tests/test_source.py +++ b/tests/test_source.py @@ -148,3 +148,17 @@ def test_facts_for_reuses_supplied_product_files_and_texts(tmp_path, monkeypatch (tmp_path / "other.py").write_text("") # the supplied text, not the disk, is what callers_of reads facts = source.facts_for(tmp_path, "app.py", 2, entry_points=[], files=["app.py", "other.py"], texts=texts) assert facts.function == "handler" and facts.callers == (("other.py", 5),) + + +def test_call_index_gives_the_same_call_sites_and_is_reused_across_names(tmp_path): + from openultrasast import source + + text = "def handler(value):\n return value\n\n\ndef run():\n x = handler(1) # handler\n return other(x)\n" + index = source.code_index(text) + assert source.call_sites(text, ("handler", "other")) == source.call_sites("", ("handler", "other"), index=index) + assert [call.line for call in source.call_sites("", ("handler",), index=index)] == [6] + (tmp_path / "other.py").write_text(text) + cache: dict = {} + assert source.callers_of(tmp_path, "handler", files=["other.py"], index=cache) == (("other.py", 6),) + (tmp_path / "other.py").write_text("") # the cached index, not the disk, answers the second name + assert source.callers_of(tmp_path, "other", files=["other.py"], index=cache) == (("other.py", 7),)