From 86c53e619ad25dfcdfb109de4593967d3d5603a6 Mon Sep 17 00:00:00 2001 From: TristanKruse Date: Sun, 5 Jul 2026 00:26:18 +0200 Subject: [PATCH] feat: resolve namespace package submodule imports --- README.md | 6 ++ .../common/extraction/extract_graph.py | 94 +++++++++++++++---- tests/common/test_extract_graph.py | 61 ++++++++++++ 3 files changed, 143 insertions(+), 18 deletions(-) diff --git a/README.md b/README.md index 9cca0bc..fcb4aa0 100644 --- a/README.md +++ b/README.md @@ -272,6 +272,12 @@ ArchUnitPython detects string-based dynamic imports such as `importlib.import_mo from my_app.adapters.sql import Repository # archunit: ignore ``` +### Namespace Packages + +ArchUnitPython resolves imports from namespace packages that do not contain +`__init__.py` files. For example, `from my_app.domain import model` is resolved +to `my_app/domain/model.py` when `my_app/domain/` is a namespace package. + ### Naming Conventions ```python diff --git a/src/archunitpython/common/extraction/extract_graph.py b/src/archunitpython/common/extraction/extract_graph.py index b28194e..7604fd0 100644 --- a/src/archunitpython/common/extraction/extract_graph.py +++ b/src/archunitpython/common/extraction/extract_graph.py @@ -41,6 +41,7 @@ class _LocatedImport: module_name: str import_kind: ImportKind line_number: int + aliases: tuple[str, ...] = () @dataclass(frozen=True) @@ -147,29 +148,28 @@ def _extract_graph_uncached( imports = _extract_located_imports(file_path) for located_import in imports: - module_name = located_import.module_name import_kind = located_import.import_kind if ( ignore_type_checking_imports and import_kind == ImportKind.TYPE_IMPORT ): continue - resolved, is_external = _resolve_import( - module_name, file_path, project_path, import_kind - ) - if resolved and resolved != _normalize(file_path): - # Check if the resolved path is in our project - if not is_external and resolved not in normalized_py_file_set: - is_external = True - - edges.append( - Edge( - source=_normalize(file_path), - target=resolved, - external=is_external, - import_kinds=(import_kind,), + for resolved, is_external in _resolve_import_targets( + located_import, file_path, project_path + ): + if resolved and resolved != _normalize(file_path): + # Check if the resolved path is in our project + if not is_external and resolved not in normalized_py_file_set: + is_external = True + + edges.append( + Edge( + source=_normalize(file_path), + target=resolved, + external=is_external, + import_kinds=(import_kind,), + ) ) - ) return _merge_edges(edges) @@ -246,11 +246,25 @@ def _extract_located_imports(file_path: str) -> list[_LocatedImport]: kind = ImportKind.TYPE_IMPORT if is_type else ImportKind.RELATIVE_IMPORT module = node.module or "" dots = "." * node.level - imports.append(_LocatedImport(f"{dots}{module}", kind, node.lineno)) + imports.append( + _LocatedImport( + f"{dots}{module}", + kind, + node.lineno, + _module_aliases(node), + ) + ) else: kind = ImportKind.TYPE_IMPORT if is_type else ImportKind.FROM_IMPORT if node.module: - imports.append(_LocatedImport(node.module, kind, node.lineno)) + imports.append( + _LocatedImport( + node.module, + kind, + node.lineno, + _module_aliases(node), + ) + ) elif isinstance(node, ast.Call): is_type = _in_type_checking(node, type_checking_ranges) @@ -321,6 +335,11 @@ def _extract_dynamic_import_names(node: ast.Call) -> list[str]: return [] +def _module_aliases(node: ast.ImportFrom) -> tuple[str, ...]: + """Return aliases that may refer to imported submodules.""" + return tuple(alias.name for alias in node.names if alias.name != "*") + + def _find_type_checking_ranges(tree: ast.Module) -> list[tuple[int, int]]: """Find line ranges of TYPE_CHECKING blocks.""" ranges: list[tuple[int, int]] = [] @@ -373,6 +392,45 @@ def _resolve_import( return _resolve_absolute_import(import_name, project_root) +def _resolve_import_targets( + import_: _LocatedImport, + source_file: str, + project_root: str, +) -> list[tuple[str, bool]]: + """Resolve an import, including namespace-package submodule aliases.""" + resolved, is_external = _resolve_import( + import_.module_name, + source_file, + project_root, + import_.import_kind, + ) + if not is_external or not import_.aliases: + return [(resolved, is_external)] + + alias_targets: list[tuple[str, bool]] = [] + for alias in import_.aliases: + alias_module = _join_import_alias(import_.module_name, alias) + alias_resolved, alias_is_external = _resolve_import( + alias_module, + source_file, + project_root, + import_.import_kind, + ) + if not alias_is_external: + alias_targets.append((alias_resolved, alias_is_external)) + + return alias_targets or [(resolved, is_external)] + + +def _join_import_alias(module_name: str, alias: str) -> str: + """Join a from-import module name with a candidate submodule alias.""" + if not module_name: + return alias + if set(module_name) == {"."}: + return f"{module_name}{alias}" + return f"{module_name}.{alias}" + + def _resolve_relative_import( import_name: str, source_file: str, diff --git a/tests/common/test_extract_graph.py b/tests/common/test_extract_graph.py index e73a072..184c699 100644 --- a/tests/common/test_extract_graph.py +++ b/tests/common/test_extract_graph.py @@ -310,6 +310,67 @@ def test_dynamic_import_resolves_to_internal_edge(self): assert ImportKind.DYNAMIC_IMPORT in edges[0].import_kinds +class TestNamespacePackageGraphHandling: + def setup_method(self): + clear_graph_cache() + + def _build_namespace_project(self, service_source: str) -> str: + temp_root = Path(__file__).resolve().parent / ".tmp" + temp_root.mkdir(exist_ok=True) + project_root = temp_root / f"project_{uuid4().hex}" + + domain_dir = project_root / "namespace_pkg" / "domain" + services_dir = project_root / "namespace_pkg" / "services" + domain_dir.mkdir(parents=True) + services_dir.mkdir(parents=True) + + (domain_dir / "model.py").write_text( + "class User:\n pass\n", + encoding="utf-8", + ) + (services_dir / "service.py").write_text(service_source, encoding="utf-8") + + self._temp_dir = project_root + return str(project_root) + + def teardown_method(self): + temp_dir = getattr(self, "_temp_dir", None) + if temp_dir is not None: + shutil.rmtree(temp_dir, ignore_errors=True) + + def _service_to_model_edges(self, project_root: str) -> list[Edge]: + graph = extract_graph(project_root) + model_path = os.path.abspath( + os.path.join(project_root, "namespace_pkg", "domain", "model.py") + ).replace("\\", "/") + service_path = os.path.abspath( + os.path.join(project_root, "namespace_pkg", "services", "service.py") + ).replace("\\", "/") + return [ + edge for edge in graph if edge.source == service_path and edge.target == model_path + ] + + def test_absolute_from_import_resolves_namespace_package_submodule(self): + project_root = self._build_namespace_project( + "from namespace_pkg.domain import model\n" + ) + + edges = self._service_to_model_edges(project_root) + + assert len(edges) == 1 + assert edges[0].external is False + assert ImportKind.FROM_IMPORT in edges[0].import_kinds + + def test_relative_from_import_resolves_namespace_package_submodule(self): + project_root = self._build_namespace_project("from ..domain import model\n") + + edges = self._service_to_model_edges(project_root) + + assert len(edges) == 1 + assert edges[0].external is False + assert ImportKind.RELATIVE_IMPORT in edges[0].import_kinds + + class TestIgnoreDirectives: def setup_method(self): clear_graph_cache()