feat: resolve namespace package submodule imports

Author: TristanKruse

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

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,

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()