diff --git a/BACKLOG.md b/BACKLOG.md index 4fefa5e..8a3af5d 100644 --- a/BACKLOG.md +++ b/BACKLOG.md @@ -18,7 +18,7 @@ This backlog collects product and maintenance ideas from project research. ## P1 - Python Import Semantics -- Add support for namespace packages that do not contain `__init__.py`. +- [x] Add support for namespace packages that do not contain `__init__.py`. - Detect dynamic imports such as `importlib.import_module()` and `__import__()`. - Detect conditional imports such as `try/except ImportError`. - Add better `TYPE_CHECKING` import handling, including options to ignore, include, or report type-only imports separately. diff --git a/README.md b/README.md index 6b8510a..6ad350b 100644 --- a/README.md +++ b/README.md @@ -345,6 +345,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 4ec3a5d..b308ea6 100644 --- a/src/archunitpython/common/extraction/extract_graph.py +++ b/src/archunitpython/common/extraction/extract_graph.py @@ -43,6 +43,7 @@ class _LocatedImport: module_name: str import_kind: ImportKind line_number: int + aliases: tuple[str, ...] = () @dataclass(frozen=True) @@ -175,29 +176,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: - continue - - 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: + continue + + edges.append( + Edge( + source=_normalize(file_path), + target=resolved, + external=is_external, + import_kinds=(import_kind,), + ) ) - ) return _merge_edges(edges) @@ -314,11 +314,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) @@ -389,6 +403,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]] = [] @@ -441,6 +460,48 @@ 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]] = [] + found_internal_alias = False + 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, + ) + alias_targets.append((alias_resolved, alias_is_external)) + found_internal_alias = found_internal_alias or not alias_is_external + + if found_internal_alias: + return alias_targets + return [(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 6372e5e..2e25e2b 100644 --- a/tests/common/test_extract_graph.py +++ b/tests/common/test_extract_graph.py @@ -378,6 +378,162 @@ 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, + *, + domain_modules: tuple[str, ...] = ("model",), + ) -> 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) + + for module in domain_modules: + (domain_dir / f"{module}.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 _service_edges(self, project_root: str) -> list[Edge]: + graph = extract_graph(project_root) + 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 != service_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 + + def test_mixed_aliases_preserve_internal_and_external_edges(self): + project_root = self._build_namespace_project( + "from namespace_pkg.domain import model, remote_model\n" + ) + + edges = self._service_edges(project_root) + model_path = os.path.abspath( + os.path.join(project_root, "namespace_pkg", "domain", "model.py") + ).replace("\\", "/") + + assert any(edge.target == model_path and not edge.external for edge in edges) + assert any( + edge.target == "namespace_pkg.domain.remote_model" and edge.external + for edge in edges + ) + + def test_multiple_internal_aliases_resolve_to_each_submodule(self): + project_root = self._build_namespace_project( + "from namespace_pkg.domain import model, audit_model\n", + domain_modules=("model", "audit_model"), + ) + + internal_targets = { + edge.target for edge in self._service_edges(project_root) if not edge.external + } + expected_targets = { + os.path.abspath( + os.path.join(project_root, "namespace_pkg", "domain", f"{module}.py") + ).replace("\\", "/") + for module in ("model", "audit_model") + } + + assert internal_targets == expected_targets + + def test_all_external_aliases_keep_original_base_edge(self): + project_root = self._build_namespace_project( + "from vendor_sdk import Client, Config\n" + ) + + external_targets = { + edge.target for edge in self._service_edges(project_root) if edge.external + } + + assert external_targets == {"vendor_sdk"} + + def test_as_alias_uses_original_submodule_name(self): + project_root = self._build_namespace_project( + "from namespace_pkg.domain import model as domain_model\n" + ) + + assert len(self._service_to_model_edges(project_root)) == 1 + + def test_relative_mixed_aliases_preserve_unresolved_edge(self): + project_root = self._build_namespace_project( + "from ..domain import model, remote_model\n" + ) + + edges = self._service_edges(project_root) + unresolved_path = os.path.abspath( + os.path.join(project_root, "namespace_pkg", "domain", "remote_model.py") + ).replace("\\", "/") + + assert len(self._service_to_model_edges(project_root)) == 1 + assert any( + edge.target == unresolved_path and edge.external for edge in edges + ) + + def test_archignore_suppresses_resolved_namespace_target(self): + project_root = self._build_namespace_project( + "from namespace_pkg.domain import model\n" + ) + Path(project_root, ".archignore").write_text( + "namespace_pkg/domain/model.py\n", + encoding="utf-8", + ) + + assert self._service_to_model_edges(project_root) == [] + + class TestIgnoreDirectives: def setup_method(self): clear_graph_cache()