deps.py
python
| 1 | """muse deps — import graph and call-graph analysis. |
| 2 | |
| 3 | Answers two questions that Git cannot: |
| 4 | |
| 5 | **File mode** (``muse deps src/billing.py``): |
| 6 | What does this file import, and what files in the repo import it? |
| 7 | |
| 8 | **Symbol mode** (``muse deps "src/billing.py::compute_invoice_total"``): |
| 9 | What does this function call? (Python only; uses stdlib ``ast``.) |
| 10 | With ``--reverse``: what symbols in the repo call this function? |
| 11 | |
| 12 | These relationships are *structural impossibilities* in Git: Git stores files |
| 13 | as blobs of text with no concept of imports or call-sites. Muse reads the |
| 14 | typed symbol graph produced at commit time and the AST of the working tree |
| 15 | to answer these questions in milliseconds. |
| 16 | |
| 17 | Usage:: |
| 18 | |
| 19 | muse deps src/billing.py # import graph (file) |
| 20 | muse deps src/billing.py --reverse # who imports this file? |
| 21 | muse deps "src/billing.py::compute_invoice_total" # call graph (Python) |
| 22 | muse deps "src/billing.py::compute_invoice_total" --reverse # callers |
| 23 | |
| 24 | Flags: |
| 25 | |
| 26 | ``--commit, -c REF`` |
| 27 | Inspect a historical snapshot instead of HEAD (import graph mode only). |
| 28 | |
| 29 | ``--reverse`` |
| 30 | Invert the query: show callers instead of callees, or importers instead |
| 31 | of imports. |
| 32 | |
| 33 | ``--json`` |
| 34 | Emit results as JSON. |
| 35 | """ |
| 36 | |
| 37 | from __future__ import annotations |
| 38 | |
| 39 | import argparse |
| 40 | import json |
| 41 | import logging |
| 42 | import pathlib |
| 43 | import sys |
| 44 | |
| 45 | from muse.core.errors import ExitCode |
| 46 | from muse.core.object_store import read_object |
| 47 | from muse.core.repo import require_repo |
| 48 | from muse.core.store import get_commit_snapshot_manifest, read_current_branch, resolve_commit_ref |
| 49 | from muse.plugins.code._callgraph import build_reverse_graph, callees_for_symbol |
| 50 | from muse.plugins.code._query import language_of, symbols_for_snapshot |
| 51 | from muse.plugins.code.ast_parser import SymbolTree, parse_symbols |
| 52 | |
| 53 | logger = logging.getLogger(__name__) |
| 54 | |
| 55 | |
| 56 | def _read_repo_id(root: pathlib.Path) -> str: |
| 57 | return str(json.loads((root / ".muse" / "repo.json").read_text())["repo_id"]) |
| 58 | |
| 59 | |
| 60 | def _read_branch(root: pathlib.Path) -> str: |
| 61 | return read_current_branch(root) |
| 62 | |
| 63 | |
| 64 | # --------------------------------------------------------------------------- |
| 65 | # Import graph helpers |
| 66 | # --------------------------------------------------------------------------- |
| 67 | |
| 68 | |
| 69 | def _imports_in_tree(tree: SymbolTree) -> list[str]: |
| 70 | """Return the list of module/symbol names imported by symbols in *tree*.""" |
| 71 | return sorted( |
| 72 | rec["qualified_name"] |
| 73 | for rec in tree.values() |
| 74 | if rec["kind"] == "import" |
| 75 | ) |
| 76 | |
| 77 | |
| 78 | def _file_imports( |
| 79 | root: pathlib.Path, |
| 80 | manifest: dict[str, str], |
| 81 | target_file: str, |
| 82 | ) -> list[str]: |
| 83 | """Return import names declared in *target_file* within *manifest*.""" |
| 84 | obj_id = manifest.get(target_file) |
| 85 | if obj_id is None: |
| 86 | return [] |
| 87 | raw = read_object(root, obj_id) |
| 88 | if raw is None: |
| 89 | return [] |
| 90 | tree = parse_symbols(raw, target_file) |
| 91 | return _imports_in_tree(tree) |
| 92 | |
| 93 | |
| 94 | def _reverse_imports( |
| 95 | root: pathlib.Path, |
| 96 | manifest: dict[str, str], |
| 97 | target_file: str, |
| 98 | ) -> list[str]: |
| 99 | """Return files in *manifest* that import a name matching *target_file*. |
| 100 | |
| 101 | The heuristic: the target file's stem (e.g. ``billing`` for |
| 102 | ``src/billing.py``) is matched against each other file's import names. |
| 103 | This catches ``import billing``, ``from billing import X``, and fully- |
| 104 | qualified paths like ``src.billing``. |
| 105 | """ |
| 106 | from muse.plugins.code.ast_parser import SEMANTIC_EXTENSIONS |
| 107 | target_stem = pathlib.PurePosixPath(target_file).stem |
| 108 | target_module = pathlib.PurePosixPath(target_file).with_suffix("").as_posix().replace("/", ".") |
| 109 | importers: list[str] = [] |
| 110 | for file_path, obj_id in manifest.items(): |
| 111 | if file_path == target_file: |
| 112 | continue |
| 113 | suffix = pathlib.PurePosixPath(file_path).suffix.lower() |
| 114 | if suffix not in SEMANTIC_EXTENSIONS: |
| 115 | continue |
| 116 | raw = read_object(root, obj_id) |
| 117 | if raw is None: |
| 118 | continue |
| 119 | tree = parse_symbols(raw, file_path) |
| 120 | for imp_name in _imports_in_tree(tree): |
| 121 | # Match stem or any suffix of the dotted module path. |
| 122 | if ( |
| 123 | imp_name == target_stem |
| 124 | or imp_name == target_module |
| 125 | or imp_name.endswith(f".{target_stem}") |
| 126 | or imp_name.endswith(f".{target_module}") |
| 127 | or target_stem in imp_name.split(".") |
| 128 | ): |
| 129 | importers.append(file_path) |
| 130 | break |
| 131 | return sorted(importers) |
| 132 | |
| 133 | |
| 134 | # --------------------------------------------------------------------------- |
| 135 | # Call-graph helpers (Python only) — thin wrappers over _callgraph |
| 136 | # --------------------------------------------------------------------------- |
| 137 | |
| 138 | |
| 139 | def _python_callers( |
| 140 | root: pathlib.Path, |
| 141 | manifest: dict[str, str], |
| 142 | target_name: str, |
| 143 | ) -> list[str]: |
| 144 | """Return addresses of Python symbols that call *target_name*.""" |
| 145 | reverse = build_reverse_graph(root, manifest) |
| 146 | return reverse.get(target_name, []) |
| 147 | |
| 148 | |
| 149 | # --------------------------------------------------------------------------- |
| 150 | # Command |
| 151 | # --------------------------------------------------------------------------- |
| 152 | |
| 153 | |
| 154 | def register(subparsers: "argparse._SubParsersAction[argparse.ArgumentParser]") -> None: |
| 155 | """Register the deps subcommand.""" |
| 156 | parser = subparsers.add_parser( |
| 157 | "deps", |
| 158 | help="Show the import graph or call graph for a file or symbol.", |
| 159 | description=__doc__, |
| 160 | ) |
| 161 | parser.add_argument( |
| 162 | "target", metavar="TARGET", |
| 163 | help=( |
| 164 | 'File path (e.g. "src/billing.py") for import graph, or ' |
| 165 | 'symbol address (e.g. "src/billing.py::compute_invoice_total") for call graph.' |
| 166 | ), |
| 167 | ) |
| 168 | parser.add_argument( |
| 169 | "--reverse", "-r", action="store_true", |
| 170 | help="Show importers (file mode) or callers (symbol mode) instead.", |
| 171 | ) |
| 172 | parser.add_argument( |
| 173 | "--commit", "-c", default=None, metavar="REF", dest="ref", |
| 174 | help="Inspect a historical commit instead of HEAD (import graph mode only).", |
| 175 | ) |
| 176 | parser.add_argument( |
| 177 | "--json", action="store_true", dest="as_json", |
| 178 | help="Emit results as JSON.", |
| 179 | ) |
| 180 | parser.set_defaults(func=run) |
| 181 | |
| 182 | |
| 183 | def run(args: argparse.Namespace) -> None: |
| 184 | """Show the import graph or call graph for a file or symbol. |
| 185 | |
| 186 | **File mode** — pass a file path:: |
| 187 | |
| 188 | muse deps src/billing.py # what does billing.py import? |
| 189 | muse deps src/billing.py --reverse # what files import billing.py? |
| 190 | |
| 191 | **Symbol mode** — pass a symbol address (Python only for call graph):: |
| 192 | |
| 193 | muse deps "src/billing.py::compute_invoice_total" |
| 194 | muse deps "src/billing.py::compute_invoice_total" --reverse |
| 195 | |
| 196 | Call-graph analysis uses the live working tree for symbol mode. |
| 197 | Import-graph analysis uses the committed snapshot (``--commit`` to pin). |
| 198 | """ |
| 199 | target: str = args.target |
| 200 | reverse: bool = args.reverse |
| 201 | ref: str | None = args.ref |
| 202 | as_json: bool = args.as_json |
| 203 | |
| 204 | root = require_repo() |
| 205 | repo_id = _read_repo_id(root) |
| 206 | branch = _read_branch(root) |
| 207 | |
| 208 | is_symbol_mode = "::" in target |
| 209 | |
| 210 | # ---------------------------------------------------------------- |
| 211 | # Symbol mode: call-graph (Python only) |
| 212 | # ---------------------------------------------------------------- |
| 213 | if is_symbol_mode: |
| 214 | file_rel, sym_qualified = target.split("::", 1) |
| 215 | lang = language_of(file_rel) |
| 216 | if lang != "Python": |
| 217 | print( |
| 218 | f"⚠️ Call-graph analysis is currently Python-only. " |
| 219 | f"'{file_rel}' is {lang}.", |
| 220 | file=sys.stderr, |
| 221 | ) |
| 222 | raise SystemExit(ExitCode.USER_ERROR) |
| 223 | |
| 224 | # Read from working tree. |
| 225 | candidates = [root / file_rel] |
| 226 | src_path: pathlib.Path | None = None |
| 227 | for c in candidates: |
| 228 | if c.exists(): |
| 229 | src_path = c |
| 230 | break |
| 231 | if src_path is None: |
| 232 | print(f"❌ File '{file_rel}' not found in working tree.", file=sys.stderr) |
| 233 | raise SystemExit(ExitCode.USER_ERROR) |
| 234 | |
| 235 | source = src_path.read_bytes() |
| 236 | |
| 237 | if not reverse: |
| 238 | callees = callees_for_symbol(source, target) |
| 239 | if as_json: |
| 240 | print(json.dumps({"address": target, "calls": callees}, indent=2)) |
| 241 | return |
| 242 | print(f"\nCallees of {target}") |
| 243 | print("─" * 62) |
| 244 | if not callees: |
| 245 | print(" (no function calls detected)") |
| 246 | else: |
| 247 | for name in callees: |
| 248 | print(f" {name}") |
| 249 | print(f"\n{len(callees)} callee(s)") |
| 250 | else: |
| 251 | target_name = sym_qualified.split(".")[-1] |
| 252 | commit = resolve_commit_ref(root, repo_id, branch, None) |
| 253 | if commit is None: |
| 254 | print("❌ No commits found.", file=sys.stderr) |
| 255 | raise SystemExit(ExitCode.USER_ERROR) |
| 256 | manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {} |
| 257 | callers = _python_callers(root, manifest, target_name) |
| 258 | if as_json: |
| 259 | print(json.dumps( |
| 260 | {"address": target, "target_name": target_name, "called_by": callers}, |
| 261 | indent=2, |
| 262 | )) |
| 263 | return |
| 264 | print(f"\nCallers of {target}") |
| 265 | print(f" (matching bare name: {target_name!r})") |
| 266 | print("─" * 62) |
| 267 | if not callers: |
| 268 | print(" (no callers found in committed snapshot)") |
| 269 | else: |
| 270 | for addr in callers: |
| 271 | print(f" {addr}") |
| 272 | print(f"\n{len(callers)} caller(s) found") |
| 273 | return |
| 274 | |
| 275 | # ---------------------------------------------------------------- |
| 276 | # File mode: import graph |
| 277 | # ---------------------------------------------------------------- |
| 278 | commit = resolve_commit_ref(root, repo_id, branch, ref) |
| 279 | if commit is None: |
| 280 | print(f"❌ Commit '{ref or 'HEAD'}' not found.", file=sys.stderr) |
| 281 | raise SystemExit(ExitCode.USER_ERROR) |
| 282 | |
| 283 | manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {} |
| 284 | |
| 285 | if not reverse: |
| 286 | imports = _file_imports(root, manifest, target) |
| 287 | if as_json: |
| 288 | print(json.dumps({"file": target, "imports": imports}, indent=2)) |
| 289 | return |
| 290 | print(f"\nImports declared in {target}") |
| 291 | print("─" * 62) |
| 292 | if not imports: |
| 293 | print(" (no imports found)") |
| 294 | else: |
| 295 | for name in imports: |
| 296 | print(f" {name}") |
| 297 | print(f"\n{len(imports)} import(s)") |
| 298 | else: |
| 299 | importers = _reverse_imports(root, manifest, target) |
| 300 | if as_json: |
| 301 | print(json.dumps({"file": target, "imported_by": importers}, indent=2)) |
| 302 | return |
| 303 | print(f"\nFiles that import {target}") |
| 304 | print("─" * 62) |
| 305 | if not importers: |
| 306 | print(" (no files import this module in the committed snapshot)") |
| 307 | else: |
| 308 | for fp in importers: |
| 309 | print(f" {fp}") |
| 310 | print(f"\n{len(importers)} importer(s) found") |