import_graph.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/deps/import_graph.py
 
"""
Purpose
-------
Build a project's dependency graph from what its modules actually
``import``, and collapse it to package level for an overview chart.
 
:mod:`oct.deps.oct_deps` builds a graph from the ``Dependencies`` section
each module docstring declares. That is a contract, and a useful one, but
a structural picture that can drift from the code is worse than none, so
the overview chart on every subproject's ``dependencies`` page is derived
from the AST instead: ``import x``, ``from x import y``, and relative
imports resolved against the importing module's package. Only imports
that resolve to a module inside the project become edges; third-party and
standard-library imports are counted, not drawn.
 
Responsibilities
----------------
- :func:`build_import_graph` - modules and edges from the AST, tests and
  the standard exclusions skipped.
- :func:`collapse_to_packages` - the package-level overview: one node per
  top-level package, edges weighted by how many module-level edges they
  stand for.
- :func:`to_mermaid_flowchart` and :func:`to_dot` - render either level.
 
Diagnostics
-----------
Domain: OCT-DEPS
Levels:
    L2 - build entry / exit with sizes
    L3 - per-module resolution decisions
    L4 - individual unresolved imports
 
Contracts
---------
- Purely static: no project module is ever imported or executed.
- A file that does not parse contributes no edges and is reported at L3;
  it never aborts the graph, because a docs build must not die on one
  broken module and the linter already blocks that module elsewhere.
- Module keys are dotted paths relative to the project root, the same
  scheme :func:`oct.deps.oct_deps.module_name_for` uses, so the two graphs
  can be compared key for key.
- ``__init__.py`` resolves to its package; an import of a package resolves
  to that package node, not to every module in it.
 
Dependencies
------------
- oct.core.diagnostics (_dbg structured logger)
- oct.deps.oct_deps (module_name_for, _should_skip - shared key scheme)
"""
 
from __future__ import annotations
 
import ast
from collections import Counter
from dataclasses import dataclass, field
from pathlib import Path
 
from oct.core.diagnostics import _dbg
from oct.deps.oct_deps import _should_skip, module_name_for
 
 
#: Directory names that hold tests. The docs tiers all leave tests out
#: (ruled 2026-09-27), and a structural overview that included them would
#: show every package depended on by "tests", which says nothing.
_TEST_DIR_NAMES: frozenset[str] = frozenset({"tests", "test"})
 
 
@dataclass
class ImportGraph:
    """The result of :func:`build_import_graph`.
 
    Attributes
    ----------
    edges
        ``module -> sorted list of project modules it imports``. Every
        module seen is a key, so isolated modules still appear as nodes.
    external
        Top-level names imported that did not resolve inside the project
        (``click``, ``json``, ...), with the number of importing modules.
    unparsed
        Project files that could not be parsed and so contributed nothing.
    """
 
    edges: dict[str, list[str]] = field(default_factory=dict)
    external: Counter[str] = field(default_factory=Counter)
    unparsed: list[Path] = field(default_factory=list)
 
    @property
    def edge_count(self) -> int:
        """Total number of module-level edges."""
        return sum(len(v) for v in self.edges.values())
 
 
#: Directories that hold Python which is not part of the project's
#: structure. ``docs`` carries the generated Sphinx ``_sphinx/conf.py``,
#: which surfaced as a lone ``docs._sphinx.conf`` node on every project's
#: module graph (found 2026-09-27 drafting plan B from this graph).
_NON_SOURCE_DIR_NAMES: frozenset[str] = frozenset({"docs"})
 
 
def _is_test_path(py_file: Path, root: Path) -> bool:
    """True when the file sits under a test or non-source directory.
 
    Only a top-level ``docs`` counts as non-source; a package that happens
    to contain a ``docs`` subpackage (``oct.docs``) is source.
    """
    try:
        parts = py_file.relative_to(root).parts[:-1]
    except ValueError:
        return False
    if parts and parts[0] in _NON_SOURCE_DIR_NAMES:
        return True
    return any(part in _TEST_DIR_NAMES for part in parts)
 
 
def _resolve_relative(module: str, is_package: bool, level: int, name: str | None) -> str | None:
    """Resolve ``from ..x import y`` to a dotted name, or ``None`` if it escapes the root.
 
    *module* is the importing module's dotted name; a package's
    ``__init__`` imports relative to itself, a plain module relative to
    its parent, which is why *is_package* matters.
    """
    base = module.split(".") if is_package else module.split(".")[:-1]
    drop = level - 1
    if drop > len(base):
        return None
    base = base[: len(base) - drop] if drop else base
    if name:
        base = base + name.split(".")
    return ".".join(base) if base else None
 
 
def _resolve_in_project(candidate: str, modules: set[str]) -> str | None:
    """Return the longest project module that *candidate* names, or ``None``.
 
    ``from oct.core import git`` may name the module ``oct.core.git`` or
    the attribute ``git`` of the package ``oct.core``; the walk from the
    longest prefix down picks whichever exists.
    """
    parts = candidate.split(".")
    for end in range(len(parts), 0, -1):
        prefix = ".".join(parts[:end])
        if prefix in modules:
            return prefix
    return None
 
 
def build_import_graph(root: Path) -> ImportGraph:
    """Walk *root* and build the import-derived module graph.
 
    Skips the standard exclusion directories and every test directory.
    See the module docstring for the resolution rules.
    """
    func = "build_import_graph"
    root = Path(root).resolve()
    _dbg("OCT-DEPS", func, f"root={root}", 2)
 
    files: dict[str, Path] = {}
    for py_file in sorted(root.rglob("*.py")):
        if _should_skip(py_file, root) or _is_test_path(py_file, root):
            continue
        files[module_name_for(py_file, root)] = py_file
    modules = set(files)
 
    graph = ImportGraph(edges={name: [] for name in files})
    for name, py_file in files.items():
        try:
            tree = ast.parse(py_file.read_text(encoding="utf-8", errors="replace"))
        except SyntaxError:
            graph.unparsed.append(py_file)
            _dbg("OCT-DEPS", func, f"unparsed: {py_file}", 3)
            continue
        is_package = py_file.name == "__init__.py"
        targets: set[str] = set()
        for node in ast.walk(tree):
            candidates: list[str] = []
            if isinstance(node, ast.Import):
                candidates = [alias.name for alias in node.names]
            elif isinstance(node, ast.ImportFrom):
                if node.level:
                    base = _resolve_relative(name, is_package, node.level, node.module)
                    if base is None:
                        continue
                    candidates = [f"{base}.{alias.name}" for alias in node.names]
                elif node.module:
                    candidates = [f"{node.module}.{alias.name}" for alias in node.names]
            for candidate in candidates:
                hit = _resolve_in_project(candidate, modules)
                if hit is None:
                    graph.external[candidate.split(".")[0]] += 1
                    _dbg("OCT-DEPS", func, f"{name}: external {candidate}", 4)
                elif hit != name:
                    targets.add(hit)
        graph.edges[name] = sorted(targets)
 
    _dbg("OCT-DEPS", func,
         f"modules={len(files)} edges={graph.edge_count} "
         f"external={len(graph.external)} unparsed={len(graph.unparsed)}", 2)
    return graph
 
 
def _package_of(module: str, depth: int) -> str:
    """The package node *module* belongs to at *depth*.
 
    The first *depth* dotted components; a name with fewer components is
    its own node, so a package's ``__init__`` (``pkg.core``) and its
    members (``pkg.core.git``) collapse to the same node. A dunder module
    (``oct.__main__``) folds into its parent rather than standing alone.
    """
    parts = module.split(".")
    if len(parts) > 1 and parts[-1].startswith("__"):
        parts = parts[:-1]
    return ".".join(parts[:depth])
 
 
def collapse_to_packages(edges: dict[str, list[str]], depth: int = 2) -> dict[tuple[str, str], int]:
    """Collapse module edges to package edges, weighted by count.
 
    *depth* is how many dotted components make a package. For a project
    laid out as ``<dist>/<pkg>/<subpkg>/module.py`` the natural overview
    is depth 2 (``oct.core``, ``oct.git``); a flat project wants 1. Edges
    inside one package are dropped - the overview is about the seams.
    """
    weights: Counter[tuple[str, str]] = Counter()
    for src, targets in edges.items():
        src_pkg = _package_of(src, depth)
        for dst in targets:
            dst_pkg = _package_of(dst, depth)
            if src_pkg != dst_pkg:
                weights[(src_pkg, dst_pkg)] += 1
    return dict(weights)
 
 
def _mermaid_id(name: str) -> str:
    return name.replace(".", "_").replace("-", "_")
 
 
def to_mermaid_flowchart(package_edges: dict[tuple[str, str], int]) -> str:
    """Render package-level edges as a Mermaid ``flowchart LR``.
 
    Edge labels carry the weight when it is more than one, so the chart
    shows where the heavy coupling is without a second diagram.
    """
    lines = ["flowchart LR"]
    nodes = sorted({n for edge in package_edges for n in edge})
    for node in nodes:
        lines.append(f'    {_mermaid_id(node)}["{node}"]')
    for (src, dst), weight in sorted(package_edges.items()):
        label = f"|{weight}|" if weight > 1 else ""
        lines.append(f"    {_mermaid_id(src)} -->{label} {_mermaid_id(dst)}")
    return "\n".join(lines)
 
 
def to_dot(edges: dict[str, list[str]], title: str = "modules") -> str:
    """Render module-level edges as Graphviz DOT, clustered by package.
 
    Clusters make the module graph readable at oct's size: dot lays each
    package out as a box, and the cross-package edges are the picture.
    """
    by_pkg: dict[str, list[str]] = {}
    for module in edges:
        by_pkg.setdefault(_package_of(module, 2), []).append(module)
    lines = [
        f'digraph "{title}" {{',
        "    rankdir=LR;",
        '    node [shape=box, fontname="Helvetica", fontsize=10];',
        '    edge [color="#666666", arrowsize=0.7];',
    ]
    for index, (pkg, members) in enumerate(sorted(by_pkg.items())):
        lines.append(f"    subgraph cluster_{index} {{")
        lines.append(f'        label="{pkg}"; style=rounded; color="#999999";')
        for module in sorted(members):
            short = module.rsplit(".", 1)[-1]
            lines.append(f'        "{module}" [label="{short}"];')
        lines.append("    }")
    for src, targets in sorted(edges.items()):
        for dst in targets:
            lines.append(f'    "{src}" -> "{dst}";')
    lines.append("}")
    return "\n".join(lines)
 
 
__all__ = [
    "ImportGraph",
    "build_import_graph",
    "collapse_to_packages",
    "to_dot",
    "to_mermaid_flowchart",
]