Speed up indexing and large architecture graph loads

src/loadpath/architecture/__init__.py

@@ -1,4 +1,4 @@
 from loadpath.architecture.rules import Finding, evaluate
-from loadpath.architecture.snapshot import architecture_report, summarize_index
+from loadpath.architecture.snapshot import architecture_report, persist_findings, summarize_index
 
-__all__ = ["Finding", "evaluate", "architecture_report", "summarize_index"]
+__all__ = ["Finding", "evaluate", "architecture_report", "summarize_index", "persist_findings"]

src/loadpath/architecture/depth.py

@@ -136,10 +136,10 @@ def _leaked_seams(store: GraphStore) -> list[Finding]:
     ]
     queries_by_src: dict[str, set[str]] = {}
     called_by_view: dict[str, set[str]] = {v: set() for v in views}
-    for edge in store.edges():
-        if edge["type"] == EdgeType.QUERIES_MODEL.value:
-            queries_by_src.setdefault(edge["src"], set()).add(edge["dst"])
-        elif edge["type"] == EdgeType.CALLS.value and edge["src"] in called_by_view:
+    for edge in store.edges_of_type([EdgeType.QUERIES_MODEL.value]):
+        queries_by_src.setdefault(edge["src"], set()).add(edge["dst"])
+    for edge in store.edges_of_type([EdgeType.CALLS.value]):
+        if edge["src"] in called_by_view:
             called_by_view[edge["src"]].add(edge["dst"])
     modules_by_ctx: dict[str, list[dict]] = {}
     for svc in services:
@@ -149,9 +149,7 @@ def _leaked_seams(store: GraphStore) -> list[Finding]:
         modules_by_ctx.setdefault(ctx, []).append(svc)
     out: list[Finding] = []
     seen: set[tuple[str, str]] = set()
-    for edge in store.edges():
-        if edge["type"] != EdgeType.QUERIES_MODEL.value:
-            continue
+    for edge in store.edges_of_type([EdgeType.QUERIES_MODEL.value]):
         view = views.get(edge["src"])
         model = models.get(edge["dst"])
         if not view or not model:
@@ -222,13 +220,12 @@ def _tests_bypass_interface(store: GraphStore) -> list[Finding]:
     views = {n["id"]: n for n in store.nodes([NodeType.VIEW])}
     serializers = {n["id"]: n for n in store.nodes([NodeType.SERIALIZER])}
     tested_src: set[str] = set()
-    for edge in store.edges():
-        if edge["type"] == EdgeType.TESTED_BY.value:
-            tested_src.add(edge["src"])
+    for edge in store.edges_of_type([EdgeType.TESTED_BY.value]):
+        tested_src.add(edge["src"])
     view_of_route: dict[str, str] = {}
     ser_of_view: dict[str, str] = {}
     page_of_route: dict[str, str] = {}
-    for edge in store.edges():
+    for edge in store.edges_of_type([EdgeType.PUBLISHES_ROUTE.value, EdgeType.USES_SERIALIZER.value]):
         if edge["type"] == EdgeType.PUBLISHES_ROUTE.value and edge["src"] in routes:
             if edge["dst"] in views:
                 view_of_route[edge["src"]] = edge["dst"]

src/loadpath/architecture/rules.py

@@ -108,9 +108,7 @@ def _views_foreign_models(store: GraphStore, config: LoadpathConfig) -> list[Fin
     out: list[Finding] = []
     views = {n["id"]: n for n in store.nodes([NodeType.VIEW, NodeType.SERVICE])}
     models = {n["id"]: n for n in store.nodes([NodeType.MODEL])}
-    for edge in store.edges():
-        if edge["type"] != EdgeType.QUERIES_MODEL.value:
-            continue
+    for edge in store.edges_of_type([EdgeType.QUERIES_MODEL.value]):
         view = views.get(edge["src"])
         model = models.get(edge["dst"]) or store.get_node(edge["dst"])
         if not view or not model:
@@ -158,9 +156,7 @@ def _react_own_api(store: GraphStore, config: LoadpathConfig) -> list[Finding]:
             allowed_by_context[name].add(path)
             allowed_by_context[name].add(normalize_url_template(path))
 
-    for edge in store.edges():
-        if edge["type"] not in {EdgeType.CALLS.value, EdgeType.CONSUMED_BY_CLIENT.value}:
-            continue
+    for edge in store.edges_of_type([EdgeType.CALLS.value, EdgeType.CONSUMED_BY_CLIENT.value]):
         src = features.get(edge["src"])
         dst = clients.get(edge["dst"])
         if not src or not dst:
@@ -214,9 +210,7 @@ def _contract_drift(store: GraphStore, config: LoadpathConfig) -> list[Finding]:
     fields = {n["id"]: n for n in store.nodes([NodeType.SERIALIZER_FIELD])}
     serializers = {n["id"]: n for n in store.nodes([NodeType.SERIALIZER])}
     matched_serializers: dict[str, list[dict]] = {}
-    for edge in store.edges():
-        if edge["type"] != EdgeType.MATCHES_SCHEMA.value:
-            continue
+    for edge in store.edges_of_type([EdgeType.MATCHES_SCHEMA.value]):
         ser = serializers.get(edge["src"])
         schema = schemas.get(edge["dst"])
         if ser and schema:
@@ -427,9 +421,7 @@ def _cascade_crosses_context(store: GraphStore, config: LoadpathConfig) -> list[
     out: list[Finding] = []
     fields = {n["id"]: n for n in store.nodes([NodeType.FIELD])}
     models = {n["id"]: n for n in store.nodes([NodeType.MODEL])}
-    for edge in store.edges():
-        if edge["type"] != EdgeType.RELATES_TO.value:
-            continue
+    for edge in store.edges_of_type([EdgeType.RELATES_TO.value]):
         if (edge.get("extra") or {}).get("on_delete") != "CASCADE":
             continue
         field = fields.get(edge["src"])
@@ -518,14 +510,12 @@ def _remaining_field_refs(store: GraphStore, app: str, model: str, field: str) -
     model_ids = _ids_for(store, NodeType.MODEL, f"{app}.{model}")
     serializer_ids: set[str] = set()
     want_model = f"{app}.{model}".lower()
-    for edge in store.edges():
-        if edge["type"] != EdgeType.SERIALIZES.value:
-            continue
+    for edge in store.edges_of_type([EdgeType.SERIALIZES.value]):
         dst = (edge.get("dst") or "").lower()
         if edge["dst"] in model_ids or dst.endswith(":" + want_model):
             serializer_ids.add(edge["src"])
-    for edge in store.edges():
-        if edge["type"] != EdgeType.HAS_FIELD.value or edge["src"] not in serializer_ids:
+    for edge in store.edges_of_type([EdgeType.HAS_FIELD.value]):
+        if edge["src"] not in serializer_ids:
             continue
         child = store.get_node(edge["dst"])
         if child and child.get("name") == field:
@@ -537,16 +527,13 @@ def _remaining_model_refs(store: GraphStore, app: str, model: str) -> list[dict]
     still = list(_qnames(store, NodeType.MODEL, f"{app}.{model}"))
     model_ids = _ids_for(store, NodeType.MODEL, f"{app}.{model}")
     want = f"{app}.{model}".lower()
-    for edge in store.edges():
+    for edge in store.edges_of_type(
+        [EdgeType.SERIALIZES.value, EdgeType.QUERIES_MODEL.value, EdgeType.RELATES_TO.value]
+    ):
         dst = (edge.get("dst") or "").lower()
         if edge["dst"] not in model_ids and not dst.endswith(":" + want):
             continue
-        if edge["type"] in {
-            EdgeType.SERIALIZES.value,
-            EdgeType.QUERIES_MODEL.value,
-            EdgeType.RELATES_TO.value,
-        }:
-            src = store.get_node(edge["src"])
-            if src:
-                still.append(src)
+        src = store.get_node(edge["src"])
+        if src:
+            still.append(src)
     return still

src/loadpath/architecture/snapshot.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import json
 from pathlib import Path
 from typing import Any
 
@@ -46,9 +47,32 @@
 }
 
 
+def persist_findings(store: GraphStore, config: LoadpathConfig) -> list[dict[str, Any]]:
+    """Run architecture rules once and store the result for cheap workspace loads."""
+    raw = evaluate(store, config)
+    findings = [f.to_dict() for f in raw]
+    store.set_meta("findings_json", json.dumps(findings))
+    return findings
+
+
+def _load_cached_findings(store: GraphStore) -> list[dict[str, Any]] | None:
+    raw = store.get_meta("findings_json")
+    if not raw:
+        return None
+    try:
+        payload = json.loads(raw)
+    except json.JSONDecodeError:
+        return None
+    if not isinstance(payload, list):
+        return None
+    return payload
+
+
 def summarize_index(store: GraphStore, config: LoadpathConfig, *, hash_drift: bool = False) -> dict[str, Any]:
-    raw_findings = evaluate(store, config)
-    findings = [f.to_dict() for f in raw_findings]
+    drift = index_drift(store, config.repo_root, config, hash_contents=hash_drift)
+    findings = None if drift.get("config_changed") else _load_cached_findings(store)
+    if findings is None:
+        findings = persist_findings(store, config)
     residuals = [line for line in (store.get_meta("residuals") or "").splitlines() if line]
     contexts = {
         name: {
@@ -60,7 +84,6 @@ def summarize_index(store: GraphStore, config: LoadpathConfig, *, hash_drift: bo
         }
         for name, ctx in config.contexts.items()
     }
-    drift = index_drift(store, config.repo_root, config, hash_contents=hash_drift)
     boot_residuals = [line for line in residuals if "django.setup()" in line]
     return {
         "ok": True,
@@ -82,7 +105,7 @@ def summarize_index(store: GraphStore, config: LoadpathConfig, *, hash_drift: bo
         "contexts": contexts,
         "rules": list(config.rules),
         "findings": findings,
-        "deepening": deepening_candidates(raw_findings),
+        "deepening": deepening_candidates(findings),
         "residuals": residuals[:40],
         "boot_residuals": boot_residuals,
         "has_config": (config.repo_root / "loadpath.yml").is_file(),

src/loadpath/detect.py

@@ -9,26 +9,9 @@
 import yaml
 
 from loadpath.config import DEFAULT_RULES, LoadpathConfig, find_config
+from loadpath.scan import SKIP_DIR_NAMES, iter_named_files, skip_dir_name
 
-SKIP_DIRS = {
-    ".git",
-    "node_modules",
-    ".venv",
-    "venv",
-    "__pycache__",
-    ".loadpath",
-    "dist",
-    "build",
-    ".mypy_cache",
-    ".pytest_cache",
-    "site-packages",
-    "docs",
-    "documentation",
-    "website",
-    "docusaurus",
-    "storybook",
-    "starlight_help",
-}
+SKIP_DIRS = SKIP_DIR_NAMES
 
 TESTISH_PARTS = {"test", "tests", "testing"}
 
@@ -41,7 +24,7 @@ def _rel_parts(path: Path, repo_root: Path) -> tuple[str, ...]:
 
 
 def _skip(path: Path, repo_root: Path) -> bool:
-    return any(part in SKIP_DIRS or part.startswith(".") for part in _rel_parts(path, repo_root))
+    return any(skip_dir_name(part) for part in _rel_parts(path, repo_root))
 
 
 def detect_layout(repo_root: Path) -> dict[str, Any]:
@@ -121,9 +104,7 @@ def ensure_config(repo_root: Path) -> LoadpathConfig:
 
 
 def _first(repo_root: Path, name: str) -> str | None:
-    for path in repo_root.rglob(name):
-        if _skip(path, repo_root):
-            continue
+    for path in iter_named_files(repo_root, (name,)):
         try:
             return path.relative_to(repo_root).as_posix()
         except ValueError:
@@ -138,9 +119,7 @@ def _is_testish(rel: Path) -> bool:
 def _detect_django_root(repo_root: Path) -> str:
     """Prefer the package that holds real apps, not a nested test project's manage.py."""
     parents: list[tuple[str, ...]] = []
-    for marker in repo_root.rglob("apps.py"):
-        if _skip(marker, repo_root):
-            continue
+    for marker in iter_named_files(repo_root, ("apps.py",)):
         app_dir = marker.parent
         if app_dir.name in {"migrations", "tests", "management"}:
             continue
@@ -157,7 +136,7 @@ def _detect_django_root(repo_root: Path) -> str:
                 break
         return "/".join(common) if common else "."
 
-    manages = [p for p in repo_root.rglob("manage.py") if not _skip(p, repo_root)]
+    manages = [p for p in iter_named_files(repo_root, ("manage.py",))]
     manages.sort(key=lambda p: (_is_testish(p.relative_to(repo_root)), len(p.relative_to(repo_root).parts)))
     if manages:
         rel = manages[0].parent.relative_to(repo_root)
@@ -223,9 +202,7 @@ def _detect_react_root(repo_root: Path) -> str:
             return candidate
 
     scored: list[tuple[int, str]] = []
-    for pkg in repo_root.rglob("package.json"):
-        if _skip(pkg, repo_root):
-            continue
+    for pkg in iter_named_files(repo_root, ("package.json",)):
         if _skip_react_tree(pkg, repo_root):
             continue
         if not _package_has_react(pkg):
@@ -250,7 +227,7 @@ def _django_apps(repo_root: Path, django_root: str) -> list[str]:
     if not root.is_dir():
         root = repo_root
     apps: list[str] = []
-    for marker in root.rglob("apps.py"):
+    for marker in iter_named_files(root, ("apps.py",)):
         if _skip(marker, repo_root):
             continue
         rel = marker.relative_to(repo_root)
@@ -260,7 +237,7 @@ def _django_apps(repo_root: Path, django_root: str) -> list[str]:
         if name not in apps and name not in {"config", "project", "settings"}:
             apps.append(name)
     if not apps:
-        for marker in root.rglob("models.py"):
+        for marker in iter_named_files(root, ("models.py",)):
             if _skip(marker, repo_root) or _is_testish(marker.relative_to(repo_root)):
                 continue
             name = marker.parent.name

src/loadpath/extractors/django.py

@@ -116,6 +116,25 @@
 ON_DELETE_ATTRS = {"CASCADE", "PROTECT", "RESTRICT", "SET_NULL", "SET_DEFAULT", "DO_NOTHING", "SET"}
 
 
+def _uppercase_name(node: ast.AST | None) -> str | None:
+    """First ClassName-looking identifier in an AST fragment (queryset = Model.objects...)."""
+    if node is None:
+        return None
+    for child in ast.walk(node):
+        if isinstance(child, ast.Name) and child.id[:1].isupper():
+            return child.id
+    return None
+
+
+def _mentions_queryset(node: ast.AST) -> bool:
+    for child in ast.walk(node):
+        if isinstance(child, ast.Name) and child.id == "queryset":
+            return True
+        if isinstance(child, ast.Attribute) and child.attr == "objects":
+            return True
+    return False
+
+
 def _name(node: ast.AST | None) -> str | None:
     if node is None:
         return None
@@ -631,7 +650,7 @@ def _serializer(
                 fname = stmt.targets[0].id
                 if not fname.startswith("_") and fname[0].islower():
                     declared.append((fname, stmt.lineno))
-            if isinstance(stmt, ast.FunctionDef) and "queryset" in ast.dump(stmt):
+            if isinstance(stmt, ast.FunctionDef) and _mentions_queryset(stmt):
                 queryset_in_serializer = True
         nested_by_name = self._nested_serializer_fields(node)
         method_fields = self._method_field_names(node)
@@ -703,10 +722,7 @@ def _view(self, node: ast.ClassDef) -> None:
                 elif key == "throttle_classes":
                     extra["throttles"] = _list_names(stmt.value)
                 elif key == "queryset":
-                    dumped = ast.dump(stmt.value)
-                    m = re.search(r"id='([A-Z][A-Za-z0-9_]+)'", dumped)
-                    if m:
-                        queryset_model = m.group(1)
+                    queryset_model = _uppercase_name(stmt.value)
                 elif key == "filterset_class":
                     extra["filterset"] = _name(stmt.value)
                 elif key == "authentication_classes":
@@ -717,10 +733,9 @@ def _view(self, node: ast.ClassDef) -> None:
                     extra["template"] = _const_str(stmt.value)
             if isinstance(stmt, ast.FunctionDef) and stmt.name == "get_queryset":
                 extra["get_queryset"] = True
-                dumped = ast.dump(stmt)
-                m = re.search(r"id='([A-Z][A-Za-z0-9_]+)'", dumped)
-                if m and not queryset_model:
-                    queryset_model = m.group(1)
+                found = _uppercase_name(stmt)
+                if found and not queryset_model:
+                    queryset_model = found
             if isinstance(stmt, ast.FunctionDef) and stmt.name == "get_serializer_class":
                 dynamic_serializer = True
                 extra["get_serializer_class"] = True

src/loadpath/extractors/django_boot.py

@@ -9,6 +9,7 @@
 from pathlib import Path
 
 from loadpath.config import LoadpathConfig
+from loadpath.scan import iter_named_files
 from loadpath.types import Edge, EdgeType, ExtractedGraph, Node, NodeType, node_id
 
 BOOT_JSON_MARKER = "__LOADPATH_BOOT_JSON__"
@@ -251,12 +252,7 @@ def _discover_settings_module(repo_root: Path, django_root: str = "backend") ->
         return env
     django_path = (repo_root / django_root).resolve()
     candidates: list[Path] = []
-    for settings in repo_root.rglob("settings.py"):
-        rel = settings.relative_to(repo_root)
-        if any(part.startswith(".") for part in rel.parts):
-            continue
-        if "site-packages" in rel.parts:
-            continue
+    for settings in iter_named_files(repo_root, ("settings.py",)):
         candidates.append(settings)
     if not candidates:
         return None