Use find_illegal_dependencies_for_layers in independence

CHANGELOG.rst

@@ -1,6 +1,12 @@
 Changelog
 =========
 
+latest
+------
+
+* Update to Grimp 3.0.
+* Use Grimp's find_illegal_dependencies_for_layers method in independence contracts.
+
 1.10.0 (2023-07-06)
 -------------------
 

pyproject.toml

@@ -13,7 +13,7 @@ authors = [
 requires-python = ">=3.8"
 dependencies = [
     "click>=6",
-    "grimp>=2.5",
+    "grimp>=3.0b3",
     "tomli>=1.2.1; python_version < '3.11'",
     "typing-extensions>=3.10.0.0",
 ]

src/importlinter/cli.py

@@ -122,7 +122,7 @@ def _combine_caching_arguments(
 
 
 def _configure_logging(verbose: bool) -> None:
-    logger_names = ("importlinter", "grimp")
+    logger_names = ("importlinter", "grimp", "_rustgrimp")
     logging_config.dictConfig(
         {
             "version": 1,

src/importlinter/contracts/_common.py

@@ -5,14 +5,17 @@
 relying on it for a custom contract type, be aware things may change
 without warning.
 """
+from __future__ import annotations
 
+import itertools
 from typing import List, Optional, Sequence, Tuple, Union
 
+import grimp
+from grimp import ImportGraph
 from typing_extensions import TypedDict
 
 from importlinter.application import output
 from importlinter.domain.imports import Module
-from grimp import ImportGraph
 
 
 class Link(TypedDict):
@@ -178,3 +181,60 @@ def _render_direct_import(
             output.print_error(f"- {import_string}", bold=False)
         else:
             output.print_error(f"  {import_string}", bold=False)
+
+
+def build_detailed_chain_from_route(route: grimp.Route, graph: grimp.ImportGraph) -> DetailedChain:
+    ordered_heads = sorted(route.heads)
+    extra_firsts: list[Link] = [
+        {
+            "importer": head,
+            "imported": route.middle[0],
+            "line_numbers": get_line_numbers(importer=head, imported=route.middle[0], graph=graph),
+        }
+        for head in ordered_heads[1:]
+    ]
+    ordered_tails = sorted(route.tails)
+    extra_lasts: list[Link] = [
+        {
+            "imported": tail,
+            "importer": route.middle[-1],
+            "line_numbers": get_line_numbers(
+                imported=tail, importer=route.middle[-1], graph=graph
+            ),
+        }
+        for tail in ordered_tails[1:]
+    ]
+    chain_as_strings = [ordered_heads[0], *route.middle, ordered_tails[0]]
+    chain_as_links: Chain = [
+        {
+            "importer": importer,
+            "imported": imported,
+            "line_numbers": get_line_numbers(importer=importer, imported=imported, graph=graph),
+        }
+        for importer, imported in pairwise(chain_as_strings)
+    ]
+    return {
+        "chain": chain_as_links,
+        "extra_firsts": extra_firsts,
+        "extra_lasts": extra_lasts,
+    }
+
+
+def get_line_numbers(
+    importer: str, imported: str, graph: grimp.ImportGraph
+) -> tuple[int | None, ...]:
+    details = graph.get_import_details(importer=importer, imported=imported)
+    line_numbers = tuple(i["line_number"] for i in details) if details else (None,)
+    return line_numbers
+
+
+def pairwise(iterable):
+    """
+    Return successive overlapping pairs taken from the input iterable.
+    pairwise('ABCDEFG') --> AB BC CD DE EF FG
+
+    TODO: Replace with itertools.pairwise once on Python 3.10.
+    """
+    a, b = itertools.tee(iterable)
+    next(b, None)
+    return zip(a, b)

src/importlinter/contracts/independence.py

@@ -1,24 +1,18 @@
-import copy
-from itertools import permutations
+from __future__ import annotations
+
 from typing import List, cast
 
+import grimp
+from grimp import ImportGraph
 from typing_extensions import TypedDict
 
 from importlinter.application import contract_utils, output
 from importlinter.application.contract_utils import AlertLevel
-from importlinter.configuration import settings
 from importlinter.domain import fields
 from importlinter.domain.contract import Contract, ContractCheck
 from importlinter.domain.imports import Module
-from grimp import ImportGraph
 
-from ._common import (
-    DetailedChain,
-    Link,
-    find_segments,
-    render_chain_data,
-    segments_to_collapsed_chains,
-)
+from ._common import DetailedChain, Link, build_detailed_chain_from_route, render_chain_data
 
 
 class _SubpackageChainData(TypedDict):
@@ -52,9 +46,6 @@ class IndependenceContract(Contract):
     unmatched_ignore_imports_alerting = fields.EnumField(AlertLevel, default=AlertLevel.ERROR)
 
     def check(self, graph: ImportGraph, verbose: bool) -> ContractCheck:
-        invalid_chains: List[_SubpackageChainData] = []
-        modules = cast(List[Module], self.modules)
-
         warnings = contract_utils.remove_ignored_imports(
             graph=graph,
             ignore_imports=self.ignore_imports,  # type: ignore
@@ -63,73 +54,14 @@ def check(self, graph: ImportGraph, verbose: bool) -> ContractCheck:
 
         self._check_all_modules_exist_in_graph(graph)
 
-        temp_graph = copy.deepcopy(graph)
-        # First pass: direct imports.
-        for subpackage_1, subpackage_2 in permutations(modules, r=2):
-            output.verbose_print(
-                verbose,
-                "Searching for direct imports from " f"{subpackage_1} to {subpackage_2}...",
-            )
-            with settings.TIMER as timer:
-                direct_chains = self._pop_direct_imports(
-                    importer_package=subpackage_1,
-                    imported_package=subpackage_2,
-                    graph=temp_graph,
-                )
-            if direct_chains:
-                invalid_chains.append(
-                    {
-                        "upstream_module": subpackage_2.name,
-                        "downstream_module": subpackage_1.name,
-                        "chains": direct_chains,
-                    }
-                )
-            if verbose:
-                chain_count = len(direct_chains)
-                pluralized = "s" if chain_count != 1 else ""
-                output.print(
-                    f"Found {chain_count} illegal chain{pluralized} "
-                    f"in {timer.duration_in_s}s.",
-                )
-
-        # Second pass: indirect imports.
-        self._squash_modules(graph=temp_graph, modules_to_squash=modules)
-        for subpackage_1, subpackage_2 in permutations(modules, r=2):
-            output.verbose_print(
-                verbose,
-                "Searching for indirect imports from " f"{subpackage_1} to {subpackage_2}...",
-            )
-            with settings.TIMER as timer:
-                other_independent_packages = [
-                    m for m in modules if m not in (subpackage_1, subpackage_2)
-                ]
-                trimmed_graph = self._make_graph_with_packages_removed(
-                    temp_graph, packages_to_remove=other_independent_packages
-                )
-                indirect_chains = self._get_indirect_collapsed_chains(
-                    trimmed_graph=trimmed_graph,
-                    reference_graph=graph,
-                    importer_package=subpackage_1,
-                    imported_package=subpackage_2,
-                )
-                if indirect_chains:
-                    invalid_chains.append(
-                        {
-                            "upstream_module": subpackage_2.name,
-                            "downstream_module": subpackage_1.name,
-                            "chains": indirect_chains,
-                        }
-                    )
-            if verbose:
-                chain_count = len(indirect_chains)
-                pluralized = "s" if chain_count != 1 else ""
-                output.print(
-                    f"Found {chain_count} illegal chain{pluralized} "
-                    f"in {timer.duration_in_s}s.",
-                )
+        dependencies = graph.find_illegal_dependencies_for_layers(
+            # A single layer consisting of siblings.
+            layers=({module.name for module in self.modules},),  # type: ignore
+        )
+        invalid_chains = self._build_invalid_chains(dependencies, graph)
 
         return ContractCheck(
-            kept=not bool(invalid_chains),
+            kept=not dependencies,
             warnings=warnings,
             metadata={"invalid_chains": invalid_chains},
         )
@@ -154,79 +86,17 @@ def _check_all_modules_exist_in_graph(self, graph: ImportGraph) -> None:
             if module.name not in graph.modules:
                 raise ValueError(f"Module '{module.name}' does not exist.")
 
-    def _squash_modules(self, graph: ImportGraph, modules_to_squash: List[Module]) -> None:
-        for module in modules_to_squash:
-            graph.squash_module(module.name)
-
-    def _make_graph_with_packages_removed(
-        self, graph: ImportGraph, packages_to_remove: List[Module]
-    ) -> ImportGraph:
-        """
-        Assumes the packages are squashed.
-        """
-        new_graph = copy.deepcopy(graph)
-        for package in packages_to_remove:
-            new_graph.remove_module(package.name)
-        return new_graph
-
-    def _get_indirect_collapsed_chains(
-        self,
-        trimmed_graph: ImportGraph,
-        reference_graph: ImportGraph,
-        importer_package: Module,
-        imported_package: Module,
-    ) -> List[DetailedChain]:
-        """
-        Return chains from the importer to the imported package.
-
-        Assumes the packages are both squashed.
-        """
-        segments = find_segments(
-            trimmed_graph,
-            reference_graph=reference_graph,
-            importer=importer_package,
-            imported=imported_package,
-        )
-        return segments_to_collapsed_chains(
-            reference_graph, segments, importer=importer_package, imported=imported_package
-        )
-
-    def _pop_direct_imports(
-        self,
-        importer_package: Module,
-        imported_package: Module,
-        graph: ImportGraph,
-    ) -> List[DetailedChain]:
-        """
-        Remove and return direct imports from the importer to the imported package.
-        """
-        direct_imports: List[DetailedChain] = []
-        importer_modules = {importer_package.name} | graph.find_descendants(importer_package.name)
-        imported_modules = {imported_package.name} | graph.find_descendants(imported_package.name)
-
-        for importer_module in importer_modules:
-            for imported_module in imported_modules:
-                import_details = graph.get_import_details(
-                    importer=importer_module, imported=imported_module
-                )
-                if import_details:
-                    line_numbers = tuple(i["line_number"] for i in import_details)
-                    direct_imports.append(
-                        {
-                            "chain": [
-                                {
-                                    "importer": import_details[0]["importer"],
-                                    "imported": import_details[0]["imported"],
-                                    "line_numbers": line_numbers,
-                                }
-                            ],
-                            "extra_firsts": [],
-                            "extra_lasts": [],
-                        }
-                    )
-                    graph.remove_import(importer=importer_module, imported=imported_module)
-
-        return direct_imports
+    def _build_invalid_chains(
+        self, dependencies: set[grimp.PackageDependency], graph: grimp.ImportGraph
+    ) -> list[_SubpackageChainData]:
+        return [
+            {
+                "upstream_module": dependency.imported,
+                "downstream_module": dependency.importer,
+                "chains": [build_detailed_chain_from_route(c, graph) for c in dependency.routes],
+            }
+            for dependency in dependencies
+        ]
 
     def _build_subpackage_chain_data(
         self, upstream_module: Module, downstream_module: Module, graph: ImportGraph