Nominate cycle breakers

CHANGELOG.rst

@@ -1,6 +1,11 @@
 Changelog
 =========
 
+latest
+------
+
+* Add --show-cycle-breakers flag.
+
 2.0 (2025-10-20)
 ----------------
 

README.rst

@@ -58,8 +58,12 @@ There is currently only one command.
     Usage: impulse drawgraph [OPTIONS] MODULE_NAME
 
     Options:
-      --show-import-totals  Label arrows with the number of imports they represent.
-      --help                Show this message and exit.
+      --show-import-totals   Label arrows with the number of imports they
+                             represent.
+      --show-cycle-breakers  Identify a set of dependencies that, if removed,
+                             would make the graph acyclic, and display them as
+                             dashed lines.
+      --help                 Show this message and exit.
 
 Draw a graph of the dependencies within any installed Python package or subpackage.
 
@@ -99,3 +103,19 @@ In this example, there is an arrow from ``.models`` to
 
 Here you can see that there are two imports from modules within ``django.db.models`` of modules
 within ``django.db.utils``.
+\
+\
+
+**Example with cycle breakers**
+
+.. code-block:: text
+
+    impulse drawgraph django.db --show-cycle-breakers
+
+.. image:: https://raw.githubusercontent.com/seddonym/impulse/master/docs/_static/images/django.db.show-cycle-breakers.png
+  :align: center
+  :alt: Graph of django.db package with cycle breakers.
+
+Here you can see that two of the dependencies are shown as a dashed line. If these dependencies were to be
+removed, the graph would be acyclic. To decide on the cycle breakers, Impulse uses the
+`nominate_cycle_breakers method provided by Grimp <https://grimp.readthedocs.io/en/stable/usage.html#ImportGraph.nominate_cycle_breakers>`_.

justfile

@@ -12,6 +12,7 @@ test:
     @uv run pytest
     @uv run impulse drawgraph grimp
     @uv run impulse drawgraph grimp --show-import-totals
+    @uv run --with=django impulse drawgraph django.db --show-cycle-breakers
 
 
 # Run tests under all supported Python versions.

pyproject.toml

@@ -9,7 +9,7 @@ authors = [
 requires-python = ">=3.9"
 dependencies = [
     "click>=6",
-    "grimp>=3.12",
+    "grimp==3.13",
     "pytest>=8.4.2",
 ]
 classifiers = [
@@ -76,9 +76,6 @@ line_length = 99
 warn_unused_ignores = true
 warn_redundant_casts = true
 
-[[tool.mypy.overrides]]
-# Ignore packages with no stubs.
-ignore_missing_imports = true
 
 [tool.uv.sources]
 import-linter = { workspace = true }

src/impulse/application/use_cases.py

@@ -1,12 +1,15 @@
+from collections.abc import Set
 from typing import Callable
 import itertools
 import grimp
 from impulse import ports, dotfile
+from typing import Optional
 
 
 def draw_graph(
     module_name: str,
     show_import_totals: bool,
+    show_cycle_breakers: bool,
     sys_path: list[str],
     current_directory: str,
     build_graph: Callable[[str], grimp.ImportGraph],
@@ -17,6 +20,7 @@ def draw_graph(
     Args:
         module_name: the package or subpackage name of any importable Python package.
         show_import_totals: whether to label the arrows with the total number of imports they represent.
+        show_cycle_breakers: marks a set of dependencies that, if removed, would make the graph acyclic.
         sys_path: the sys.path list (or a test double).
         current_directory: the current working directory.
         build_graph: the function which builds the graph of the supplied package
@@ -27,36 +31,165 @@ def draw_graph(
     sys_path.insert(0, current_directory)
 
     module = grimp.Module(module_name)
-    graph = build_graph(module.package_name)
-    module_children = graph.find_children(module.name)
+    grimp_graph = build_graph(module.package_name)
 
-    dot = dotfile.DotGraph(title=module_name)
+    dot = _build_dot(grimp_graph, module_name, show_import_totals, show_cycle_breakers)
 
-    for module_child in module_children:
-        dot.add_node(module_child)
+    viewer.view(dot)
+
+
+class _DotGraphBuildStrategy:
+    def build(self, module_name: str, grimp_graph: grimp.ImportGraph) -> dotfile.DotGraph:
+        children = grimp_graph.find_children(module_name)
+
+        self.prepare_graph(grimp_graph, children)
+
+        dot = dotfile.DotGraph(title=module_name, concentrate=self.should_concentrate())
+        for child in children:
+            dot.add_node(child)
+        for upstream, downstream in itertools.permutations(children, r=2):
+            if edge := self.build_edge(grimp_graph, upstream, downstream):
+                dot.add_edge(edge)
+
+        return dot
+
+    def should_concentrate(self) -> bool:
+        return True
+
+    def prepare_graph(self, grimp_graph: grimp.ImportGraph, children: Set[str]) -> None:
+        pass
+
+    def build_edge(
+        self, grimp_graph: grimp.ImportGraph, upstream: str, downstream: str
+    ) -> Optional[dotfile.Edge]:
+        raise NotImplementedError
+
+
+class _ModuleSquashingBuildStrategy(_DotGraphBuildStrategy):
+    """Fast builder for when we don't need additional data about the imports."""
+
+    def prepare_graph(self, grimp_graph: grimp.ImportGraph, children: Set[str]) -> None:
+        for child in children:
+            grimp_graph.squash_module(child)
+
+    def build_edge(
+        self, grimp_graph: grimp.ImportGraph, upstream: str, downstream: str
+    ) -> Optional[dotfile.Edge]:
+        if grimp_graph.direct_import_exists(importer=downstream, imported=upstream):
+            return dotfile.Edge(source=downstream, destination=upstream)
+        return None
+
+
+class _ImportExpressionBuildStrategy(_DotGraphBuildStrategy):
+    """Slower builder for when we want to work on the whole graph,
+    without squashing children.
+    """
+
+    def __init__(
+        self, *, module_name: str, show_import_totals: bool, show_cycle_breakers: bool
+    ) -> None:
+        self.module_name = module_name
+        self.show_import_totals = show_import_totals
+        self.show_cycle_breakers = show_cycle_breakers
+        self.cycle_breakers: Optional[set[tuple[str, str]]] = None
+
+    def should_concentrate(self) -> bool:
+        # We need to see edge direction emphasized separately.
+        return not (self.show_import_totals or self.show_cycle_breakers)
+
+    def prepare_graph(self, grimp_graph: grimp.ImportGraph, children: Set[str]) -> None:
+        super().prepare_graph(grimp_graph, children)
 
-    # Dependencies between children.
-    for upstream, downstream in itertools.permutations(module_children, r=2):
-        if graph.direct_import_exists(importer=downstream, imported=upstream, as_packages=True):
-            if show_import_totals:
-                number_of_imports = _count_imports_between_packages(
-                    graph, importer=downstream, imported=upstream
+        if self.show_cycle_breakers:
+            self.cycle_breakers = self._get_coarse_grained_cycle_breakers(grimp_graph, children)
+
+    def _get_coarse_grained_cycle_breakers(
+        self, grimp_graph: grimp.ImportGraph, children: Set[str]
+    ) -> set[tuple[str, str]]:
+        # In the form (importer, imported).
+        coarse_grained_cycle_breakers: set[tuple[str, str]] = set()
+
+        for fine_grained_cycle_breaker in grimp_graph.nominate_cycle_breakers(self.module_name):
+            importer, imported = fine_grained_cycle_breaker
+            importer_ancestor = self._get_self_or_ancestor(candidate=importer, ancestors=children)
+            imported_ancestor = self._get_self_or_ancestor(candidate=imported, ancestors=children)
+
+            if importer_ancestor and imported_ancestor:
+                coarse_grained_cycle_breakers.add((importer_ancestor, imported_ancestor))
+
+        return coarse_grained_cycle_breakers
+
+    @staticmethod
+    def _get_self_or_ancestor(candidate: str, ancestors: Set[str]) -> Optional[str]:
+        for ancestor in ancestors:
+            if candidate == ancestor or candidate.startswith(f"{ancestor}."):
+                return ancestor
+        return None
+
+    def build_edge(
+        self, grimp_graph: grimp.ImportGraph, upstream: str, downstream: str
+    ) -> Optional[dotfile.Edge]:
+        if grimp_graph.direct_import_exists(
+            importer=downstream, imported=upstream, as_packages=True
+        ):
+            if self.show_import_totals:
+                number_of_imports = self._count_imports_between_packages(
+                    grimp_graph, importer=downstream, imported=upstream
                 )
                 label = str(number_of_imports)
             else:
                 label = ""
-            dot.add_edge(source=downstream, destination=upstream, label=label)
-    viewer.view(dot)
 
+            if self.show_cycle_breakers:
+                assert self.cycle_breakers is not None
+                is_cycle_breaker = (downstream, upstream) in self.cycle_breakers
+                emphasized = is_cycle_breaker
+            else:
+                emphasized = False
+
+            return dotfile.Edge(
+                source=downstream, destination=upstream, label=label, emphasized=emphasized
+            )
+        return None
 
-def _count_imports_between_packages(
-    graph: grimp.ImportGraph, *, importer: str, imported: str
-) -> int:
-    return (
-        len(graph.find_matching_direct_imports(import_expression=f"{importer} -> {imported}"))
-        + len(graph.find_matching_direct_imports(import_expression=f"{importer} -> {imported}.**"))
-        + len(graph.find_matching_direct_imports(import_expression=f"{importer}.** -> {imported}"))
-        + len(
-            graph.find_matching_direct_imports(import_expression=f"{importer}.** -> {imported}.**")
+    @staticmethod
+    def _count_imports_between_packages(
+        graph: grimp.ImportGraph, *, importer: str, imported: str
+    ) -> int:
+        return (
+            len(graph.find_matching_direct_imports(import_expression=f"{importer} -> {imported}"))
+            + len(
+                graph.find_matching_direct_imports(
+                    import_expression=f"{importer} -> {imported}.**"
+                )
+            )
+            + len(
+                graph.find_matching_direct_imports(
+                    import_expression=f"{importer}.** -> {imported}"
+                )
+            )
+            + len(
+                graph.find_matching_direct_imports(
+                    import_expression=f"{importer}.** -> {imported}.**"
+                )
+            )
         )
-    )
+
+
+def _build_dot(
+    grimp_graph: grimp.ImportGraph,
+    module_name: str,
+    show_import_totals: bool,
+    show_cycle_breakers: bool,
+) -> dotfile.DotGraph:
+    strategy: _DotGraphBuildStrategy
+    if show_import_totals or show_cycle_breakers:
+        strategy = _ImportExpressionBuildStrategy(
+            module_name=module_name,
+            show_import_totals=show_import_totals,
+            show_cycle_breakers=show_cycle_breakers,
+        )
+    else:
+        strategy = _ModuleSquashingBuildStrategy()
+
+    return strategy.build(module_name, grimp_graph)

src/impulse/cli.py

@@ -19,11 +19,20 @@ def main():
     is_flag=True,
     help="Label arrows with the number of imports they represent.",
 )
+@click.option(
+    "--show-cycle-breakers",
+    is_flag=True,
+    help=(
+        "Identify a set of dependencies that, if removed, would make the graph acyclic, "
+        "and display them as dashed lines."
+    ),
+)
 @click.argument("module_name", type=str)
-def drawgraph(module_name: str, show_import_totals: bool) -> None:
+def drawgraph(module_name: str, show_import_totals: bool, show_cycle_breakers: bool) -> None:
     use_cases.draw_graph(
         module_name=module_name,
         show_import_totals=show_import_totals,
+        show_cycle_breakers=show_cycle_breakers,
         sys_path=sys.path,
         current_directory=os.getcwd(),
         build_graph=grimp.build_graph,

src/impulse/dotfile.py

@@ -1,4 +1,28 @@
 from textwrap import dedent
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True, order=True)
+class Edge:
+    source: str
+    destination: str
+    label: str = ""
+    emphasized: bool = False
+
+    def __str__(self) -> str:
+        return f'"{DotGraph.render_module(self.source)}" ->  "{DotGraph.render_module(self.destination)}"{self._render_attrs()}\n'
+
+    def _render_attrs(self) -> str:
+        attrs: dict[str, str] = {}
+        if self.label:
+            attrs["label"] = self.label
+        if self.emphasized:
+            attrs["style"] = "dashed"
+        if attrs:
+            joined_attrs = ", ".join([f'{key}="{value}"' for key, value in attrs.items()])
+            return f" [{joined_attrs}]"
+        else:
+            return ""
 
 
 class DotGraph:
@@ -8,37 +32,34 @@ class DotGraph:
     https://en.wikipedia.org/wiki/DOT_(graph_description_language)
     """
 
-    def __init__(self, title: str) -> None:
+    def __init__(self, title: str, concentrate: bool = True) -> None:
         self.title = title
         self.nodes: set[str] = set()
-        self.edges: set[tuple[str, str, str]] = set()
+        self.edges: set[Edge] = set()
+        self.concentrate = concentrate
 
     def add_node(self, name: str) -> None:
         self.nodes.add(name)
 
-    def add_edge(self, *, source: str, destination: str, label: str) -> None:
-        self.edges.add((source, destination, label))
+    def add_edge(self, edge: Edge) -> None:
+        self.edges.add(edge)
 
     def render(self) -> str:
         # concentrate=true means that we merge the lines together.
         return dedent(f"""digraph {{
             node [fontname=helvetica]
-            concentrate=true
+            {"concentrate=true" if self.concentrate else ""}
             {self._render_nodes()}
             {self._render_edges()}
         }}""")
 
     def _render_nodes(self) -> str:
-        return "\n".join(f'"{self._render_module(node)}"\n' for node in sorted(self.nodes))
+        return "\n".join(f'"{self.render_module(node)}"\n' for node in sorted(self.nodes))
 
     def _render_edges(self) -> str:
-        return "\n".join(
-            f'"{self._render_module(source)}" ->  "{self._render_module(destination)}"{self._render_label(label)}\n'
-            for source, destination, label in sorted(self.edges)
-        )
-
-    def _render_module(self, module: str) -> str:
-        return module.rsplit(self.title)[1]
+        return "\n".join(str(edge) for edge in sorted(self.edges))
 
-    def _render_label(self, label: str) -> str:
-        return f" [label={label}]" if label else ""
+    @staticmethod
+    def render_module(module: str) -> str:
+        # Render as relative module.
+        return f".{module.split('.')[-1]}"