Run all post-build steps on a populated graph. Each step is non-fatal: failures are logged and collected as warnings so the primary build result is never lost. Args: store: An open GraphStore with nodes and edges already populated. Returns: Dict with keys for each
(store: GraphStore)
| 24 | |
| 25 | |
| 26 | def run_post_processing(store: GraphStore) -> dict[str, Any]: |
| 27 | """Run all post-build steps on a populated graph. |
| 28 | |
| 29 | Each step is non-fatal: failures are logged and collected as warnings |
| 30 | so the primary build result is never lost. |
| 31 | |
| 32 | Args: |
| 33 | store: An open GraphStore with nodes and edges already populated. |
| 34 | |
| 35 | Returns: |
| 36 | Dict with keys for each step's result count and a ``warnings`` |
| 37 | list (only present when at least one step failed). |
| 38 | """ |
| 39 | result: dict[str, Any] = {} |
| 40 | warnings: list[str] = [] |
| 41 | |
| 42 | _compute_signatures(store, result, warnings) |
| 43 | _rebuild_fts_index(store, result, warnings) |
| 44 | _trace_flows(store, result, warnings) |
| 45 | _detect_communities(store, result, warnings) |
| 46 | |
| 47 | if warnings: |
| 48 | result["warnings"] = warnings |
| 49 | return result |
| 50 | |
| 51 | |
| 52 | # -- Individual steps (private) ------------------------------------------ |