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 json |
| 40 | import logging |
| 41 | import pathlib |
| 42 | |
| 43 | import typer |
| 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 | app = typer.Typer() |
| 56 | |
| 57 | |
| 58 | def _read_repo_id(root: pathlib.Path) -> str: |
| 59 | return str(json.loads((root / ".muse" / "repo.json").read_text())["repo_id"]) |
| 60 | |
| 61 | |
| 62 | def _read_branch(root: pathlib.Path) -> str: |
| 63 | return read_current_branch(root) |
| 64 | |
| 65 | |
| 66 | # --------------------------------------------------------------------------- |
| 67 | # Import graph helpers |
| 68 | # --------------------------------------------------------------------------- |
| 69 | |
| 70 | |
| 71 | def _imports_in_tree(tree: SymbolTree) -> list[str]: |
| 72 | """Return the list of module/symbol names imported by symbols in *tree*.""" |
| 73 | return sorted( |
| 74 | rec["qualified_name"] |
| 75 | for rec in tree.values() |
| 76 | if rec["kind"] == "import" |
| 77 | ) |
| 78 | |
| 79 | |
| 80 | def _file_imports( |
| 81 | root: pathlib.Path, |
| 82 | manifest: dict[str, str], |
| 83 | target_file: str, |
| 84 | ) -> list[str]: |
| 85 | """Return import names declared in *target_file* within *manifest*.""" |
| 86 | obj_id = manifest.get(target_file) |
| 87 | if obj_id is None: |
| 88 | return [] |
| 89 | raw = read_object(root, obj_id) |
| 90 | if raw is None: |
| 91 | return [] |
| 92 | tree = parse_symbols(raw, target_file) |
| 93 | return _imports_in_tree(tree) |
| 94 | |
| 95 | |
| 96 | def _reverse_imports( |
| 97 | root: pathlib.Path, |
| 98 | manifest: dict[str, str], |
| 99 | target_file: str, |
| 100 | ) -> list[str]: |
| 101 | """Return files in *manifest* that import a name matching *target_file*. |
| 102 | |
| 103 | The heuristic: the target file's stem (e.g. ``billing`` for |
| 104 | ``src/billing.py``) is matched against each other file's import names. |
| 105 | This catches ``import billing``, ``from billing import X``, and fully- |
| 106 | qualified paths like ``src.billing``. |
| 107 | """ |
| 108 | from muse.plugins.code.ast_parser import SEMANTIC_EXTENSIONS |
| 109 | target_stem = pathlib.PurePosixPath(target_file).stem |
| 110 | target_module = pathlib.PurePosixPath(target_file).with_suffix("").as_posix().replace("/", ".") |
| 111 | importers: list[str] = [] |
| 112 | for file_path, obj_id in manifest.items(): |
| 113 | if file_path == target_file: |
| 114 | continue |
| 115 | suffix = pathlib.PurePosixPath(file_path).suffix.lower() |
| 116 | if suffix not in SEMANTIC_EXTENSIONS: |
| 117 | continue |
| 118 | raw = read_object(root, obj_id) |
| 119 | if raw is None: |
| 120 | continue |
| 121 | tree = parse_symbols(raw, file_path) |
| 122 | for imp_name in _imports_in_tree(tree): |
| 123 | # Match stem or any suffix of the dotted module path. |
| 124 | if ( |
| 125 | imp_name == target_stem |
| 126 | or imp_name == target_module |
| 127 | or imp_name.endswith(f".{target_stem}") |
| 128 | or imp_name.endswith(f".{target_module}") |
| 129 | or target_stem in imp_name.split(".") |
| 130 | ): |
| 131 | importers.append(file_path) |
| 132 | break |
| 133 | return sorted(importers) |
| 134 | |
| 135 | |
| 136 | # --------------------------------------------------------------------------- |
| 137 | # Call-graph helpers (Python only) — thin wrappers over _callgraph |
| 138 | # --------------------------------------------------------------------------- |
| 139 | |
| 140 | |
| 141 | def _python_callers( |
| 142 | root: pathlib.Path, |
| 143 | manifest: dict[str, str], |
| 144 | target_name: str, |
| 145 | ) -> list[str]: |
| 146 | """Return addresses of Python symbols that call *target_name*.""" |
| 147 | reverse = build_reverse_graph(root, manifest) |
| 148 | return reverse.get(target_name, []) |
| 149 | |
| 150 | |
| 151 | # --------------------------------------------------------------------------- |
| 152 | # Command |
| 153 | # --------------------------------------------------------------------------- |
| 154 | |
| 155 | |
| 156 | @app.callback(invoke_without_command=True) |
| 157 | def deps( |
| 158 | ctx: typer.Context, |
| 159 | target: str = typer.Argument( |
| 160 | ..., metavar="TARGET", |
| 161 | help=( |
| 162 | 'File path (e.g. "src/billing.py") for import graph, or ' |
| 163 | 'symbol address (e.g. "src/billing.py::compute_invoice_total") for call graph.' |
| 164 | ), |
| 165 | ), |
| 166 | reverse: bool = typer.Option( |
| 167 | False, "--reverse", "-r", |
| 168 | help="Show importers (file mode) or callers (symbol mode) instead.", |
| 169 | ), |
| 170 | ref: str | None = typer.Option( |
| 171 | None, "--commit", "-c", metavar="REF", |
| 172 | help="Inspect a historical commit instead of HEAD (import graph mode only).", |
| 173 | ), |
| 174 | as_json: bool = typer.Option(False, "--json", help="Emit results as JSON."), |
| 175 | ) -> None: |
| 176 | """Show the import graph or call graph for a file or symbol. |
| 177 | |
| 178 | **File mode** — pass a file path:: |
| 179 | |
| 180 | muse deps src/billing.py # what does billing.py import? |
| 181 | muse deps src/billing.py --reverse # what files import billing.py? |
| 182 | |
| 183 | **Symbol mode** — pass a symbol address (Python only for call graph):: |
| 184 | |
| 185 | muse deps "src/billing.py::compute_invoice_total" |
| 186 | muse deps "src/billing.py::compute_invoice_total" --reverse |
| 187 | |
| 188 | Call-graph analysis uses the live working tree for symbol mode. |
| 189 | Import-graph analysis uses the committed snapshot (``--commit`` to pin). |
| 190 | """ |
| 191 | root = require_repo() |
| 192 | repo_id = _read_repo_id(root) |
| 193 | branch = _read_branch(root) |
| 194 | |
| 195 | is_symbol_mode = "::" in target |
| 196 | |
| 197 | # ---------------------------------------------------------------- |
| 198 | # Symbol mode: call-graph (Python only) |
| 199 | # ---------------------------------------------------------------- |
| 200 | if is_symbol_mode: |
| 201 | file_rel, sym_qualified = target.split("::", 1) |
| 202 | lang = language_of(file_rel) |
| 203 | if lang != "Python": |
| 204 | typer.echo( |
| 205 | f"⚠️ Call-graph analysis is currently Python-only. " |
| 206 | f"'{file_rel}' is {lang}.", |
| 207 | err=True, |
| 208 | ) |
| 209 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 210 | |
| 211 | # Read from working tree. |
| 212 | candidates = [root / file_rel] |
| 213 | src_path: pathlib.Path | None = None |
| 214 | for c in candidates: |
| 215 | if c.exists(): |
| 216 | src_path = c |
| 217 | break |
| 218 | if src_path is None: |
| 219 | typer.echo(f"❌ File '{file_rel}' not found in working tree.", err=True) |
| 220 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 221 | |
| 222 | source = src_path.read_bytes() |
| 223 | |
| 224 | if not reverse: |
| 225 | callees = callees_for_symbol(source, target) |
| 226 | if as_json: |
| 227 | typer.echo(json.dumps({"address": target, "calls": callees}, indent=2)) |
| 228 | return |
| 229 | typer.echo(f"\nCallees of {target}") |
| 230 | typer.echo("─" * 62) |
| 231 | if not callees: |
| 232 | typer.echo(" (no function calls detected)") |
| 233 | else: |
| 234 | for name in callees: |
| 235 | typer.echo(f" {name}") |
| 236 | typer.echo(f"\n{len(callees)} callee(s)") |
| 237 | else: |
| 238 | target_name = sym_qualified.split(".")[-1] |
| 239 | commit = resolve_commit_ref(root, repo_id, branch, None) |
| 240 | if commit is None: |
| 241 | typer.echo("❌ No commits found.", err=True) |
| 242 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 243 | manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {} |
| 244 | callers = _python_callers(root, manifest, target_name) |
| 245 | if as_json: |
| 246 | typer.echo(json.dumps( |
| 247 | {"address": target, "target_name": target_name, "called_by": callers}, |
| 248 | indent=2, |
| 249 | )) |
| 250 | return |
| 251 | typer.echo(f"\nCallers of {target}") |
| 252 | typer.echo(f" (matching bare name: {target_name!r})") |
| 253 | typer.echo("─" * 62) |
| 254 | if not callers: |
| 255 | typer.echo(" (no callers found in committed snapshot)") |
| 256 | else: |
| 257 | for addr in callers: |
| 258 | typer.echo(f" {addr}") |
| 259 | typer.echo(f"\n{len(callers)} caller(s) found") |
| 260 | return |
| 261 | |
| 262 | # ---------------------------------------------------------------- |
| 263 | # File mode: import graph |
| 264 | # ---------------------------------------------------------------- |
| 265 | commit = resolve_commit_ref(root, repo_id, branch, ref) |
| 266 | if commit is None: |
| 267 | typer.echo(f"❌ Commit '{ref or 'HEAD'}' not found.", err=True) |
| 268 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 269 | |
| 270 | manifest = get_commit_snapshot_manifest(root, commit.commit_id) or {} |
| 271 | |
| 272 | if not reverse: |
| 273 | imports = _file_imports(root, manifest, target) |
| 274 | if as_json: |
| 275 | typer.echo(json.dumps({"file": target, "imports": imports}, indent=2)) |
| 276 | return |
| 277 | typer.echo(f"\nImports declared in {target}") |
| 278 | typer.echo("─" * 62) |
| 279 | if not imports: |
| 280 | typer.echo(" (no imports found)") |
| 281 | else: |
| 282 | for name in imports: |
| 283 | typer.echo(f" {name}") |
| 284 | typer.echo(f"\n{len(imports)} import(s)") |
| 285 | else: |
| 286 | importers = _reverse_imports(root, manifest, target) |
| 287 | if as_json: |
| 288 | typer.echo(json.dumps({"file": target, "imported_by": importers}, indent=2)) |
| 289 | return |
| 290 | typer.echo(f"\nFiles that import {target}") |
| 291 | typer.echo("─" * 62) |
| 292 | if not importers: |
| 293 | typer.echo(" (no files import this module in the committed snapshot)") |
| 294 | else: |
| 295 | for fp in importers: |
| 296 | typer.echo(f" {fp}") |
| 297 | typer.echo(f"\n{len(importers)} importer(s) found") |