_framework.py
python
sha256:502e812e12319e30302d07d165425ceeb21687ad30e8af3a439b1f6de36c64d6
fix: update stale tag tests to use label for free-form anno…
Sonnet 4.6
19 days ago
| 1 | """Framework entry-point detection for the code-domain symbol graph. |
| 2 | |
| 3 | Problem |
| 4 | ------- |
| 5 | Static call-graph analysis only records *explicit* call edges — ``A()`` inside |
| 6 | ``B`` produces edge ``B→A``. Framework entry points (FastAPI route handlers, |
| 7 | Flask views, Celery tasks, …) are wired at *runtime* through decorators and DI |
| 8 | containers. No explicit call site exists in user code, so they appear dead to |
| 9 | :func:`~muse.cli.commands.dead.run` and have empty blast-radius in |
| 10 | :func:`~muse.cli.commands.impact.run`. |
| 11 | |
| 12 | Solution |
| 13 | -------- |
| 14 | This module adds a complementary edge type — :class:`ImplicitEntryEdge` — that |
| 15 | represents "this symbol is externally reachable via the framework named |
| 16 | ``framework_id``." Edges are synthesised by :class:`FrameworkPlugin` |
| 17 | implementations, one per framework, during a single linear pass over the |
| 18 | manifest. The resulting :data:`ImplicitEdgeGraph` is consumed by the impact |
| 19 | and dead-code commands to: |
| 20 | |
| 21 | * Surface entry-point annotations instead of "no callers detected". |
| 22 | * Exclude entry-point symbols from dead-code results. |
| 23 | |
| 24 | Adding a new framework |
| 25 | ---------------------- |
| 26 | 1. Create a class that satisfies the :class:`FrameworkPlugin` protocol. |
| 27 | 2. Append an instance to :data:`BUILTIN_PLUGINS`. |
| 28 | 3. Write tests in ``tests/test_framework_plugins.py``. |
| 29 | |
| 30 | No changes to the core graph, impact, or dead-code modules are required. |
| 31 | |
| 32 | Configuration |
| 33 | ------------- |
| 34 | ``.muse/code_config.toml`` is the single configuration file for all code-domain |
| 35 | intelligence. It has two independent sections: |
| 36 | |
| 37 | ``[code]`` — controls which tree-sitter language grammars are active:: |
| 38 | |
| 39 | [code] |
| 40 | # Restrict to Python + TypeScript + shell; all other grammars become |
| 41 | # file-level fallbacks even if the grammar packages are installed. |
| 42 | languages = ["python", "typescript", "tsx", "css", "bash"] |
| 43 | |
| 44 | ``[framework_detection]`` — controls entry-point detection:: |
| 45 | |
| 46 | [framework_detection] |
| 47 | auto_detect = true |
| 48 | disabled_plugins = [] # e.g. ["celery"] to silence false positives |
| 49 | |
| 50 | [[framework_detection.custom_entry_points]] |
| 51 | language = "Python" |
| 52 | kind = "custom-rpc" |
| 53 | decorator_names = ["rpc_handler", "grpc_handler"] |
| 54 | |
| 55 | Both sections are optional; omitting either reproduces the zero-config defaults. |
| 56 | """ |
| 57 | |
| 58 | import ast |
| 59 | import logging |
| 60 | import pathlib |
| 61 | import tomllib |
| 62 | from dataclasses import dataclass, field |
| 63 | from typing import Protocol, runtime_checkable |
| 64 | |
| 65 | from muse.core.types import Manifest |
| 66 | from muse.core.paths import code_config_path as _code_config_path |
| 67 | from muse.core.object_store import read_object |
| 68 | from muse.core.symbol_cache import SymbolCache |
| 69 | from muse.core.implicit_edge_cache import ImplicitEdgeCache |
| 70 | from muse.core.validation import MAX_AST_BYTES |
| 71 | from muse.plugins.code.ast_parser import SymbolTree, parse_symbols |
| 72 | |
| 73 | logger = logging.getLogger(__name__) |
| 74 | |
| 75 | # --------------------------------------------------------------------------- |
| 76 | # Core data types |
| 77 | # --------------------------------------------------------------------------- |
| 78 | |
| 79 | #: String-keyed metadata dict for per-framework entry-point annotations. |
| 80 | #: Keys vary by framework (e.g. "method", "path" for HTTP; "hook" for Flask). |
| 81 | type _Metadata = dict[str, str] |
| 82 | |
| 83 | @dataclass(frozen=True, slots=True) |
| 84 | class ImplicitEntryEdge: |
| 85 | """One framework-implied reachability edge for a symbol. |
| 86 | |
| 87 | Attributes: |
| 88 | framework_id: Stable lowercase identifier, e.g. ``"fastapi"``. |
| 89 | symbol_address: Full symbol address, e.g. |
| 90 | ``"app/routers/runs.py::create_run"``. |
| 91 | kind: Semantic category of the entry point: |
| 92 | ``"http-route"``, ``"task"``, ``"cli-command"``, … |
| 93 | metadata: Arbitrary per-framework annotations. Common keys: |
| 94 | ``method`` (HTTP verb), ``path`` (URL pattern), |
| 95 | ``queue`` (Celery queue name). |
| 96 | """ |
| 97 | |
| 98 | framework_id: str |
| 99 | symbol_address: str |
| 100 | kind: str |
| 101 | metadata: _Metadata = field(default_factory=dict) |
| 102 | |
| 103 | #: Maps ``symbol_address → [ImplicitEntryEdge, …]`` for every framework-wired |
| 104 | #: symbol in a snapshot. Analogous to :data:`~._callgraph.ReverseGraph` but |
| 105 | #: for implicit framework callers rather than explicit call sites. |
| 106 | ImplicitEdgeGraph = dict[str, list[ImplicitEntryEdge]] |
| 107 | |
| 108 | # --------------------------------------------------------------------------- |
| 109 | # FrameworkPlugin protocol |
| 110 | # --------------------------------------------------------------------------- |
| 111 | |
| 112 | @runtime_checkable |
| 113 | class FrameworkPlugin(Protocol): |
| 114 | """Protocol every framework entry-point plugin must satisfy. |
| 115 | |
| 116 | Implementations should be stateless singletons. The contract is: |
| 117 | |
| 118 | * ``language`` — the primary language this plugin targets. Used to skip |
| 119 | files whose suffix does not match, avoiding unnecessary parses. |
| 120 | * ``framework_id`` — a stable lowercase slug, e.g. ``"fastapi"``. |
| 121 | * ``detect_entry_points`` — pure, thread-safe, returns an empty list when |
| 122 | the file contains no framework-wired symbols. |
| 123 | """ |
| 124 | |
| 125 | @property |
| 126 | def language(self) -> str: |
| 127 | """Primary language, e.g. ``"Python"`` or ``"TypeScript"``.""" |
| 128 | ... |
| 129 | |
| 130 | @property |
| 131 | def framework_id(self) -> str: |
| 132 | """Stable lowercase slug, e.g. ``"fastapi"``.""" |
| 133 | ... |
| 134 | |
| 135 | def detect_entry_points( |
| 136 | self, |
| 137 | file_path: str, |
| 138 | sym_tree: SymbolTree, |
| 139 | source: bytes, |
| 140 | ) -> list[ImplicitEntryEdge]: |
| 141 | """Return implicit entry edges for framework-wired symbols in *file_path*. |
| 142 | |
| 143 | Args: |
| 144 | file_path: Workspace-relative POSIX path, e.g. |
| 145 | ``"app/routers/runs.py"``. |
| 146 | sym_tree: Pre-parsed symbol tree for the file. Plugins may |
| 147 | use this to correlate decorator findings with known |
| 148 | symbol addresses without re-parsing. |
| 149 | source: Raw file bytes. Plugins that need full AST detail |
| 150 | (e.g. decorator arguments) must call |
| 151 | ``ast.parse(source)`` themselves. |
| 152 | |
| 153 | Returns: |
| 154 | A (possibly empty) list of :class:`ImplicitEntryEdge` objects. |
| 155 | One edge per framework-wired symbol per entry-point type. |
| 156 | """ |
| 157 | ... |
| 158 | |
| 159 | # --------------------------------------------------------------------------- |
| 160 | # Configuration |
| 161 | # --------------------------------------------------------------------------- |
| 162 | |
| 163 | @dataclass |
| 164 | class _CustomEntryPointRule: |
| 165 | """One user-defined entry-point pattern from ``.muse/code_config.toml``.""" |
| 166 | |
| 167 | language: str |
| 168 | kind: str |
| 169 | decorator_names: list[str] = field(default_factory=list) |
| 170 | |
| 171 | @dataclass |
| 172 | class FrameworkConfig: |
| 173 | """Parsed ``[framework_detection]`` section from ``.muse/code_config.toml``. |
| 174 | |
| 175 | Defaults reflect the "zero-config" happy path: all built-in plugins active, |
| 176 | no custom rules. |
| 177 | """ |
| 178 | |
| 179 | auto_detect: bool = True |
| 180 | disabled_plugins: frozenset[str] = field(default_factory=frozenset) |
| 181 | custom_entry_points: list[_CustomEntryPointRule] = field(default_factory=list) |
| 182 | |
| 183 | def load_framework_config(root: pathlib.Path) -> FrameworkConfig: |
| 184 | """Read ``[framework_detection]`` from ``.muse/code_config.toml``. |
| 185 | |
| 186 | Returns default :class:`FrameworkConfig` if the file is absent or the |
| 187 | section is missing — callers should not need to handle ``None``. |
| 188 | """ |
| 189 | config_path = _code_config_path(root) |
| 190 | if not config_path.exists(): |
| 191 | return FrameworkConfig() |
| 192 | |
| 193 | try: |
| 194 | with open(config_path, "rb") as fh: |
| 195 | raw = tomllib.load(fh) |
| 196 | except Exception as exc: # noqa: BLE001 |
| 197 | logger.warning("⚠️ Could not parse .muse/code_config.toml: %s", exc) |
| 198 | return FrameworkConfig() |
| 199 | |
| 200 | section = raw.get("framework_detection", {}) |
| 201 | if not isinstance(section, dict): |
| 202 | return FrameworkConfig() |
| 203 | |
| 204 | auto_detect: bool = bool(section.get("auto_detect", True)) |
| 205 | disabled: list[str] = section.get("disabled_plugins", []) |
| 206 | custom: list[_CustomEntryPointRule] = [] |
| 207 | for rule in section.get("custom_entry_points", []): |
| 208 | if not isinstance(rule, dict): |
| 209 | continue |
| 210 | lang = rule.get("language", "Python") |
| 211 | kind = rule.get("kind", "custom") |
| 212 | names = rule.get("decorator_names", []) |
| 213 | if isinstance(names, list): |
| 214 | custom.append(_CustomEntryPointRule(language=lang, kind=kind, decorator_names=names)) |
| 215 | |
| 216 | return FrameworkConfig( |
| 217 | auto_detect=auto_detect, |
| 218 | disabled_plugins=frozenset(disabled), |
| 219 | custom_entry_points=custom, |
| 220 | ) |
| 221 | |
| 222 | # --------------------------------------------------------------------------- |
| 223 | # Custom pattern plugin (reads user-defined rules from FrameworkConfig) |
| 224 | # --------------------------------------------------------------------------- |
| 225 | |
| 226 | class _CustomPatternPlugin: |
| 227 | """Synthesises entry edges from ``[[framework_detection.custom_entry_points]]`` rules.""" |
| 228 | |
| 229 | language = "Python" |
| 230 | framework_id = "custom" |
| 231 | |
| 232 | def __init__(self, rules: list[_CustomEntryPointRule]) -> None: |
| 233 | self._rules = [r for r in rules if r.language == "Python"] |
| 234 | |
| 235 | def detect_entry_points( |
| 236 | self, |
| 237 | file_path: str, |
| 238 | sym_tree: SymbolTree, |
| 239 | source: bytes, |
| 240 | ) -> list[ImplicitEntryEdge]: |
| 241 | if not self._rules or len(source) > MAX_AST_BYTES: |
| 242 | return [] |
| 243 | try: |
| 244 | tree = ast.parse(source) |
| 245 | except SyntaxError: |
| 246 | return [] |
| 247 | |
| 248 | decorator_to_kind = {} # bare_name → kind string |
| 249 | for rule in self._rules: |
| 250 | for name in rule.decorator_names: |
| 251 | decorator_to_kind[name] = rule.kind |
| 252 | |
| 253 | if not decorator_to_kind: |
| 254 | return [] |
| 255 | |
| 256 | edges: list[ImplicitEntryEdge] = [] |
| 257 | for node in ast.walk(tree): |
| 258 | if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): |
| 259 | continue |
| 260 | for dec in node.decorator_list: |
| 261 | dec_name = _decorator_bare_name(dec) |
| 262 | if dec_name and dec_name in decorator_to_kind: |
| 263 | addr = _resolve_address(file_path, node.name, sym_tree) |
| 264 | if addr: |
| 265 | edges.append(ImplicitEntryEdge( |
| 266 | framework_id=self.framework_id, |
| 267 | symbol_address=addr, |
| 268 | kind=decorator_to_kind[dec_name], |
| 269 | metadata={"decorator": dec_name}, |
| 270 | )) |
| 271 | return edges |
| 272 | |
| 273 | # --------------------------------------------------------------------------- |
| 274 | # Built-in plugins |
| 275 | # --------------------------------------------------------------------------- |
| 276 | |
| 277 | class _FastAPIPlugin: |
| 278 | """Detects FastAPI / APIRouter route handlers. |
| 279 | |
| 280 | Recognises the decorator forms:: |
| 281 | |
| 282 | @app.get("/path") |
| 283 | @router.post("/path", ...) |
| 284 | @api_router.put("/path") |
| 285 | |
| 286 | Any variable name is accepted as the router/app object; only the HTTP |
| 287 | *method attribute* is checked (``get``, ``post``, ``put``, ``delete``, |
| 288 | ``patch``, ``head``, ``options``, ``trace``). |
| 289 | """ |
| 290 | |
| 291 | language = "Python" |
| 292 | framework_id = "fastapi" |
| 293 | |
| 294 | _HTTP_METHODS: frozenset[str] = frozenset( |
| 295 | {"get", "post", "put", "delete", "patch", "head", "options", "trace"} |
| 296 | ) |
| 297 | |
| 298 | def detect_entry_points( |
| 299 | self, |
| 300 | file_path: str, |
| 301 | sym_tree: SymbolTree, |
| 302 | source: bytes, |
| 303 | ) -> list[ImplicitEntryEdge]: |
| 304 | if len(source) > MAX_AST_BYTES: |
| 305 | return [] |
| 306 | try: |
| 307 | tree = ast.parse(source) |
| 308 | except SyntaxError: |
| 309 | return [] |
| 310 | |
| 311 | edges: list[ImplicitEntryEdge] = [] |
| 312 | for node in ast.walk(tree): |
| 313 | if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): |
| 314 | continue |
| 315 | for dec in node.decorator_list: |
| 316 | edge = self._edge_from_decorator(dec, node.name, file_path, sym_tree) |
| 317 | if edge is not None: |
| 318 | edges.append(edge) |
| 319 | return edges |
| 320 | |
| 321 | def _edge_from_decorator( |
| 322 | self, |
| 323 | dec: ast.expr, |
| 324 | func_name: str, |
| 325 | file_path: str, |
| 326 | sym_tree: SymbolTree, |
| 327 | ) -> ImplicitEntryEdge | None: |
| 328 | # Must be a Call: @router.get("/path") or @router.get("/path", ...) |
| 329 | if not isinstance(dec, ast.Call): |
| 330 | return None |
| 331 | func = dec.func |
| 332 | # Must be an attribute: <anything>.get / .post / etc. |
| 333 | if not isinstance(func, ast.Attribute): |
| 334 | return None |
| 335 | method = func.attr.lower() |
| 336 | if method not in self._HTTP_METHODS: |
| 337 | return None |
| 338 | |
| 339 | # Extract the path from the first positional argument (optional). |
| 340 | path = "" |
| 341 | if dec.args and isinstance(dec.args[0], ast.Constant): |
| 342 | path = str(dec.args[0].value) |
| 343 | |
| 344 | addr = _resolve_address(file_path, func_name, sym_tree) |
| 345 | if addr is None: |
| 346 | return None |
| 347 | |
| 348 | return ImplicitEntryEdge( |
| 349 | framework_id=self.framework_id, |
| 350 | symbol_address=addr, |
| 351 | kind="http-route", |
| 352 | metadata={"method": method.upper(), "path": path}, |
| 353 | ) |
| 354 | |
| 355 | class _FlaskPlugin: |
| 356 | """Detects Flask and Blueprint route handlers and lifecycle hooks. |
| 357 | |
| 358 | Recognises:: |
| 359 | |
| 360 | @app.route("/path", methods=["GET", "POST"]) |
| 361 | @bp.route("/path") |
| 362 | @app.before_request / @app.after_request / @app.teardown_appcontext |
| 363 | @app.errorhandler(404) |
| 364 | @app.cli.command() |
| 365 | """ |
| 366 | |
| 367 | language = "Python" |
| 368 | framework_id = "flask" |
| 369 | |
| 370 | _ROUTE_ATTRS: frozenset[str] = frozenset({"route"}) |
| 371 | _LIFECYCLE_ATTRS: frozenset[str] = frozenset( |
| 372 | {"before_request", "after_request", "teardown_appcontext", |
| 373 | "before_app_request", "after_app_request"} |
| 374 | ) |
| 375 | _ERROR_ATTRS: frozenset[str] = frozenset({"errorhandler"}) |
| 376 | |
| 377 | def detect_entry_points( |
| 378 | self, |
| 379 | file_path: str, |
| 380 | sym_tree: SymbolTree, |
| 381 | source: bytes, |
| 382 | ) -> list[ImplicitEntryEdge]: |
| 383 | if len(source) > MAX_AST_BYTES: |
| 384 | return [] |
| 385 | try: |
| 386 | tree = ast.parse(source) |
| 387 | except SyntaxError: |
| 388 | return [] |
| 389 | |
| 390 | edges: list[ImplicitEntryEdge] = [] |
| 391 | for node in ast.walk(tree): |
| 392 | if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): |
| 393 | continue |
| 394 | for dec in node.decorator_list: |
| 395 | edge = self._edge_from_decorator(dec, node.name, file_path, sym_tree) |
| 396 | if edge is not None: |
| 397 | edges.append(edge) |
| 398 | return edges |
| 399 | |
| 400 | def _edge_from_decorator( |
| 401 | self, |
| 402 | dec: ast.expr, |
| 403 | func_name: str, |
| 404 | file_path: str, |
| 405 | sym_tree: SymbolTree, |
| 406 | ) -> ImplicitEntryEdge | None: |
| 407 | addr = _resolve_address(file_path, func_name, sym_tree) |
| 408 | if addr is None: |
| 409 | return None |
| 410 | |
| 411 | # @app.route("/path", methods=[...]) |
| 412 | if isinstance(dec, ast.Call) and isinstance(dec.func, ast.Attribute): |
| 413 | attr = dec.func.attr |
| 414 | if attr in self._ROUTE_ATTRS: |
| 415 | path = "" |
| 416 | if dec.args and isinstance(dec.args[0], ast.Constant): |
| 417 | path = str(dec.args[0].value) |
| 418 | methods = _extract_methods_kwarg(dec) or ["GET"] |
| 419 | return ImplicitEntryEdge( |
| 420 | framework_id=self.framework_id, |
| 421 | symbol_address=addr, |
| 422 | kind="http-route", |
| 423 | metadata={"method": ",".join(methods), "path": path}, |
| 424 | ) |
| 425 | if attr in self._LIFECYCLE_ATTRS: |
| 426 | return ImplicitEntryEdge( |
| 427 | framework_id=self.framework_id, |
| 428 | symbol_address=addr, |
| 429 | kind="lifecycle-hook", |
| 430 | metadata={"hook": attr}, |
| 431 | ) |
| 432 | if attr in self._ERROR_ATTRS: |
| 433 | return ImplicitEntryEdge( |
| 434 | framework_id=self.framework_id, |
| 435 | symbol_address=addr, |
| 436 | kind="error-handler", |
| 437 | metadata={}, |
| 438 | ) |
| 439 | |
| 440 | # Bare @app.before_request (no call parens) |
| 441 | if isinstance(dec, ast.Attribute) and dec.attr in self._LIFECYCLE_ATTRS: |
| 442 | return ImplicitEntryEdge( |
| 443 | framework_id=self.framework_id, |
| 444 | symbol_address=addr, |
| 445 | kind="lifecycle-hook", |
| 446 | metadata={"hook": dec.attr}, |
| 447 | ) |
| 448 | |
| 449 | return None |
| 450 | |
| 451 | class _CeleryPlugin: |
| 452 | """Detects Celery task functions. |
| 453 | |
| 454 | Recognises:: |
| 455 | |
| 456 | @app.task |
| 457 | @celery.task |
| 458 | @shared_task |
| 459 | @app.task(bind=True) |
| 460 | @app.periodic_task(run_every=…) |
| 461 | """ |
| 462 | |
| 463 | language = "Python" |
| 464 | framework_id = "celery" |
| 465 | |
| 466 | _TASK_BARE: frozenset[str] = frozenset({"shared_task"}) |
| 467 | _TASK_ATTRS: frozenset[str] = frozenset({"task", "periodic_task"}) |
| 468 | |
| 469 | def detect_entry_points( |
| 470 | self, |
| 471 | file_path: str, |
| 472 | sym_tree: SymbolTree, |
| 473 | source: bytes, |
| 474 | ) -> list[ImplicitEntryEdge]: |
| 475 | if len(source) > MAX_AST_BYTES: |
| 476 | return [] |
| 477 | try: |
| 478 | tree = ast.parse(source) |
| 479 | except SyntaxError: |
| 480 | return [] |
| 481 | |
| 482 | edges: list[ImplicitEntryEdge] = [] |
| 483 | for node in ast.walk(tree): |
| 484 | if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): |
| 485 | continue |
| 486 | for dec in node.decorator_list: |
| 487 | edge = self._edge_from_decorator(dec, node.name, file_path, sym_tree) |
| 488 | if edge is not None: |
| 489 | edges.append(edge) |
| 490 | return edges |
| 491 | |
| 492 | def _edge_from_decorator( |
| 493 | self, |
| 494 | dec: ast.expr, |
| 495 | func_name: str, |
| 496 | file_path: str, |
| 497 | sym_tree: SymbolTree, |
| 498 | ) -> ImplicitEntryEdge | None: |
| 499 | addr = _resolve_address(file_path, func_name, sym_tree) |
| 500 | if addr is None: |
| 501 | return None |
| 502 | |
| 503 | # @shared_task (bare Name) |
| 504 | if isinstance(dec, ast.Name) and dec.id in self._TASK_BARE: |
| 505 | return ImplicitEntryEdge( |
| 506 | framework_id=self.framework_id, |
| 507 | symbol_address=addr, |
| 508 | kind="task", |
| 509 | metadata={}, |
| 510 | ) |
| 511 | |
| 512 | # @shared_task(...) (called Name) |
| 513 | if isinstance(dec, ast.Call) and isinstance(dec.func, ast.Name): |
| 514 | if dec.func.id in self._TASK_BARE: |
| 515 | return ImplicitEntryEdge( |
| 516 | framework_id=self.framework_id, |
| 517 | symbol_address=addr, |
| 518 | kind="task", |
| 519 | metadata={}, |
| 520 | ) |
| 521 | |
| 522 | # @app.task / @celery.task (bare Attribute) |
| 523 | if isinstance(dec, ast.Attribute) and dec.attr in self._TASK_ATTRS: |
| 524 | kind = "periodic-task" if dec.attr == "periodic_task" else "task" |
| 525 | return ImplicitEntryEdge( |
| 526 | framework_id=self.framework_id, |
| 527 | symbol_address=addr, |
| 528 | kind=kind, |
| 529 | metadata={}, |
| 530 | ) |
| 531 | |
| 532 | # @app.task(...) / @app.periodic_task(...) (called Attribute) |
| 533 | if isinstance(dec, ast.Call) and isinstance(dec.func, ast.Attribute): |
| 534 | if dec.func.attr in self._TASK_ATTRS: |
| 535 | kind = "periodic-task" if dec.func.attr == "periodic_task" else "task" |
| 536 | return ImplicitEntryEdge( |
| 537 | framework_id=self.framework_id, |
| 538 | symbol_address=addr, |
| 539 | kind=kind, |
| 540 | metadata={}, |
| 541 | ) |
| 542 | |
| 543 | return None |
| 544 | |
| 545 | #: Default plugin instances — extend this list to add new frameworks. |
| 546 | BUILTIN_PLUGINS: list[FrameworkPlugin] = [ |
| 547 | _FastAPIPlugin(), |
| 548 | _FlaskPlugin(), |
| 549 | _CeleryPlugin(), |
| 550 | ] |
| 551 | |
| 552 | # --------------------------------------------------------------------------- |
| 553 | # Graph construction |
| 554 | # --------------------------------------------------------------------------- |
| 555 | |
| 556 | _PY_SUFFIXES: frozenset[str] = frozenset({".py", ".pyi"}) |
| 557 | |
| 558 | def build_implicit_edge_graph( |
| 559 | root: pathlib.Path, |
| 560 | manifest: Manifest, |
| 561 | cache: SymbolCache | None = None, |
| 562 | config: FrameworkConfig | None = None, |
| 563 | implicit_cache: ImplicitEdgeCache | None = None, |
| 564 | ) -> ImplicitEdgeGraph: |
| 565 | """Build the implicit entry-edge graph from *manifest*. |
| 566 | |
| 567 | Runs all active :data:`BUILTIN_PLUGINS` (plus any custom rules from |
| 568 | *config*) over every file in the manifest and aggregates the results. |
| 569 | |
| 570 | Two levels of caching are supported: |
| 571 | |
| 572 | * ``implicit_cache`` — fastest path: skips ``read_object``, symbol |
| 573 | parsing, and all plugin ``detect_entry_points`` calls for files |
| 574 | whose edge list is already cached. |
| 575 | * ``cache`` (``SymbolCache``) — skips ``parse_symbols`` on a hit, but |
| 576 | ``read_object`` and plugin detection are still required. |
| 577 | |
| 578 | The caller is responsible for calling ``implicit_cache.save()`` after |
| 579 | the build if persistence is desired. |
| 580 | |
| 581 | Args: |
| 582 | root: Repository root — used to read blobs from the object store. |
| 583 | manifest: Snapshot manifest mapping file path → SHA-256 object ID. |
| 584 | cache: Optional :class:`~muse.core.symbol_cache.SymbolCache`. |
| 585 | config: Framework configuration. When ``None``, |
| 586 | :func:`load_framework_config` is called with *root*. |
| 587 | implicit_cache: Optional :class:`~muse.core.implicit_edge_cache.ImplicitEdgeCache`. |
| 588 | When provided and a file's edge list is cached, the file |
| 589 | is skipped entirely — no blob fetch, no plugin run. |
| 590 | |
| 591 | Returns: |
| 592 | :data:`ImplicitEdgeGraph` — ``{symbol_address: [ImplicitEntryEdge, …]}``. |
| 593 | Symbols with no implicit callers are absent from the mapping. |
| 594 | """ |
| 595 | if config is None: |
| 596 | config = load_framework_config(root) |
| 597 | |
| 598 | if not config.auto_detect: |
| 599 | return {} |
| 600 | |
| 601 | active_plugins: list[FrameworkPlugin] = [ |
| 602 | p for p in BUILTIN_PLUGINS |
| 603 | if p.framework_id not in config.disabled_plugins |
| 604 | ] |
| 605 | if config.custom_entry_points: |
| 606 | active_plugins.append(_CustomPatternPlugin(config.custom_entry_points)) |
| 607 | |
| 608 | if not active_plugins: |
| 609 | return {} |
| 610 | |
| 611 | # Partition plugins by language so we only run Python plugins on .py files. |
| 612 | py_plugins = [p for p in active_plugins if p.language == "Python"] |
| 613 | |
| 614 | graph: ImplicitEdgeGraph = {} |
| 615 | |
| 616 | for file_path, obj_id in manifest.items(): |
| 617 | suffix = pathlib.PurePosixPath(file_path).suffix.lower() |
| 618 | |
| 619 | plugins_for_file = py_plugins if suffix in _PY_SUFFIXES else [] |
| 620 | if not plugins_for_file: |
| 621 | continue |
| 622 | |
| 623 | # Fast path: return cached edge list — no read_object, no plugin run. |
| 624 | if implicit_cache is not None: |
| 625 | cached = implicit_cache.get(obj_id) |
| 626 | if cached is not None: |
| 627 | for edge in cached: |
| 628 | graph.setdefault(edge.symbol_address, []).append(edge) |
| 629 | continue |
| 630 | |
| 631 | raw = read_object(root, obj_id) |
| 632 | if raw is None or len(raw) > MAX_AST_BYTES: |
| 633 | continue |
| 634 | |
| 635 | sym_tree: SymbolTree = ( |
| 636 | cache.get(obj_id) or parse_symbols(raw, file_path) |
| 637 | if cache is not None |
| 638 | else parse_symbols(raw, file_path) |
| 639 | ) |
| 640 | |
| 641 | file_edges: list[ImplicitEntryEdge] = [] |
| 642 | for plugin in plugins_for_file: |
| 643 | try: |
| 644 | edges = plugin.detect_entry_points(file_path, sym_tree, raw) |
| 645 | except Exception as exc: # noqa: BLE001 |
| 646 | logger.warning( |
| 647 | "⚠️ %s plugin raised on %s: %s", |
| 648 | plugin.framework_id, file_path, exc, |
| 649 | ) |
| 650 | continue |
| 651 | file_edges.extend(edges) |
| 652 | |
| 653 | if implicit_cache is not None: |
| 654 | implicit_cache.put(obj_id, file_edges) |
| 655 | |
| 656 | for edge in file_edges: |
| 657 | graph.setdefault(edge.symbol_address, []).append(edge) |
| 658 | |
| 659 | return graph |
| 660 | |
| 661 | # --------------------------------------------------------------------------- |
| 662 | # Internal helpers |
| 663 | # --------------------------------------------------------------------------- |
| 664 | |
| 665 | def _resolve_address( |
| 666 | file_path: str, |
| 667 | func_name: str, |
| 668 | sym_tree: SymbolTree, |
| 669 | ) -> str | None: |
| 670 | """Return the full symbol address for *func_name* within *file_path*. |
| 671 | |
| 672 | Tries exact match first (``file::name``), then a scan for method forms |
| 673 | (``file::ClassName.name``). Returns ``None`` if the name is not in the |
| 674 | symbol tree — this prevents synthesising edges for lambdas or inline |
| 675 | functions that the AST parser does not index. |
| 676 | """ |
| 677 | direct = f"{file_path}::{func_name}" |
| 678 | if direct in sym_tree: |
| 679 | return direct |
| 680 | # Method match: "file::SomeClass.func_name" |
| 681 | for addr in sym_tree: |
| 682 | if addr.endswith(f".{func_name}") and addr.startswith(f"{file_path}::"): |
| 683 | return addr |
| 684 | return None |
| 685 | |
| 686 | def _decorator_bare_name(dec: ast.expr) -> str | None: |
| 687 | """Return the bare decorator name for simple ``@name`` or ``@name(...)`` forms.""" |
| 688 | if isinstance(dec, ast.Name): |
| 689 | return dec.id |
| 690 | if isinstance(dec, ast.Call) and isinstance(dec.func, ast.Name): |
| 691 | return dec.func.id |
| 692 | if isinstance(dec, ast.Attribute): |
| 693 | return dec.attr |
| 694 | if isinstance(dec, ast.Call) and isinstance(dec.func, ast.Attribute): |
| 695 | return dec.func.attr |
| 696 | return None |
| 697 | |
| 698 | def _extract_methods_kwarg(call_node: ast.Call) -> list[str] | None: |
| 699 | """Extract HTTP methods from a ``methods=[...]`` keyword in a decorator call.""" |
| 700 | for kw in call_node.keywords: |
| 701 | if kw.arg == "methods" and isinstance(kw.value, ast.List): |
| 702 | methods: list[str] = [] |
| 703 | for elt in kw.value.elts: |
| 704 | if isinstance(elt, ast.Constant): |
| 705 | methods.append(str(elt.value).upper()) |
| 706 | return methods or None |
| 707 | return None |
File History
1 commit
sha256:502e812e12319e30302d07d165425ceeb21687ad30e8af3a439b1f6de36c64d6
fix: update stale tag tests to use label for free-form anno…
Sonnet 4.6
19 days ago