diff --git a/BACKLOG.md b/BACKLOG.md index 09a5593..f4c0a11 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`. - [x] Detect dynamic imports such as `importlib.import_module()` and `__import__()`. - [x] 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 0c4c70c..efeb3ae 100644 --- a/README.md +++ b/README.md @@ -344,6 +344,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. + ### Conditional Imports Imports inside `try` blocks that handle `ImportError` or diff --git a/src/archunitpython/common/extraction/extract_graph.py b/src/archunitpython/common/extraction/extract_graph.py index 0e085f8..e0ddc14 100644 --- a/src/archunitpython/common/extraction/extract_graph.py +++ b/src/archunitpython/common/extraction/extract_graph.py @@ -45,6 +45,7 @@ class _LocatedImport: line_number: int resolution_kind: ImportKind | None = None fallback_module_name: str | None = None + aliases: tuple[str, ...] = () @dataclass(frozen=True) @@ -177,39 +178,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 - resolution_kind = located_import.resolution_kind or import_kind - resolved, is_external = _resolve_import( - module_name, file_path, project_path, resolution_kind - ) - if is_external and located_import.fallback_module_name is not None: - fallback, fallback_is_external = _resolve_import( - located_import.fallback_module_name, - file_path, - project_path, - resolution_kind, - ) - if fallback and not fallback_is_external: - resolved, is_external = fallback, False - 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=_edge_import_kinds(located_import), + 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=_edge_import_kinds(located_import), + ) ) - ) return _merge_edges(edges) @@ -342,14 +332,16 @@ def _extract_located_imports(file_path: str) -> list[_LocatedImport]: fallback_module_name = ( "." * node.level if node.level and node.module is None else None ) + aliases = _module_aliases(node) if node.module else () for module_name in _import_from_module_names(node): imports.append( _LocatedImport( - module_name, - kind, - node.lineno, - syntax_kind, - fallback_module_name, + module_name=module_name, + import_kind=kind, + line_number=node.lineno, + resolution_kind=syntax_kind, + fallback_module_name=fallback_module_name, + aliases=aliases, ) ) @@ -429,6 +421,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 _import_from_module_names(node: ast.ImportFrom) -> tuple[str, ...]: """Return resolvable module names for a from-import statement.""" dots = "." * (node.level or 0) @@ -573,6 +570,58 @@ 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.""" + resolution_kind = import_.resolution_kind or import_.import_kind + resolved, is_external = _resolve_import( + import_.module_name, + source_file, + project_root, + resolution_kind, + ) + if is_external and import_.fallback_module_name is not None: + fallback, fallback_is_external = _resolve_import( + import_.fallback_module_name, + source_file, + project_root, + resolution_kind, + ) + if fallback and not fallback_is_external: + resolved, is_external = fallback, False + 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, + resolution_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 1436e7b..f0b25cb 100644 --- a/tests/common/test_extract_graph.py +++ b/tests/common/test_extract_graph.py @@ -404,6 +404,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 TestConditionalImportGraphHandling: def setup_method(self): clear_graph_cache()