diff --git a/BACKLOG.md b/BACKLOG.md index 4fefa5e..9fc4fb7 100644 --- a/BACKLOG.md +++ b/BACKLOG.md @@ -20,7 +20,7 @@ This backlog collects product and maintenance ideas from project research. - 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`. +- [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. - Improve external dependency rules so users can express allowed and forbidden third-party packages at module or slice level. - Consider a public-interface rule inspired by Tach, where modules may only import through declared package APIs. diff --git a/README.md b/README.md index 6b8510a..72eb858 100644 --- a/README.md +++ b/README.md @@ -345,6 +345,15 @@ ArchUnitPython detects string-based dynamic imports such as `importlib.import_mo from my_app.adapters.sql import Repository # archunit: ignore ``` +### Conditional Imports + +Imports inside `try` blocks that handle `ImportError` or +`ModuleNotFoundError` are marked as conditional dependencies. This helps graph +reports distinguish optional imports and fallback implementations from regular +runtime imports. Conditional dependencies remain part of architecture checks; +relative and dynamic imports also retain their original import kind in graph +reports. + ### Naming Conventions ```python diff --git a/src/archunitpython/common/extraction/extract_graph.py b/src/archunitpython/common/extraction/extract_graph.py index 4ec3a5d..2deffaf 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 + resolution_kind: ImportKind | None = None @dataclass(frozen=True) @@ -182,8 +183,9 @@ def _extract_graph_uncached( 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, import_kind + module_name, file_path, project_path, resolution_kind ) if resolved and resolved != _normalize(file_path): # Check if the resolved path is in our project @@ -195,7 +197,7 @@ def _extract_graph_uncached( source=_normalize(file_path), target=resolved, external=is_external, - import_kinds=(import_kind,), + import_kinds=_edge_import_kinds(located_import), ) ) @@ -299,32 +301,56 @@ def _extract_located_imports(file_path: str) -> list[_LocatedImport]: imports: list[_LocatedImport] = [] ignore_directives = _find_ignore_directives(source) type_checking_ranges = _find_type_checking_ranges(tree) + conditional_import_ranges = _find_conditional_import_ranges(tree) for node in ast.walk(tree): if isinstance(node, ast.Import): - is_type = _in_type_checking(node, type_checking_ranges) - kind = ImportKind.TYPE_IMPORT if is_type else ImportKind.IMPORT + syntax_kind = ImportKind.IMPORT + kind = _classify_import( + node, + syntax_kind, + type_checking_ranges, + conditional_import_ranges, + ) for alias in node.names: - imports.append(_LocatedImport(alias.name, kind, node.lineno)) + imports.append( + _LocatedImport(alias.name, kind, node.lineno, syntax_kind) + ) elif isinstance(node, ast.ImportFrom): - is_type = _in_type_checking(node, type_checking_ranges) - if node.level and node.level > 0: - # Relative import - 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)) - else: - kind = ImportKind.TYPE_IMPORT if is_type else ImportKind.FROM_IMPORT - if node.module: - imports.append(_LocatedImport(node.module, kind, node.lineno)) + syntax_kind = ( + ImportKind.RELATIVE_IMPORT + if node.level and node.level > 0 + else ImportKind.FROM_IMPORT + ) + kind = _classify_import( + node, + syntax_kind, + type_checking_ranges, + conditional_import_ranges, + ) + for module_name in _import_from_module_names(node): + imports.append( + _LocatedImport( + module_name, + kind, + node.lineno, + syntax_kind, + ) + ) elif isinstance(node, ast.Call): - is_type = _in_type_checking(node, type_checking_ranges) - kind = ImportKind.TYPE_IMPORT if is_type else ImportKind.DYNAMIC_IMPORT + syntax_kind = ImportKind.DYNAMIC_IMPORT + kind = _classify_import( + node, + syntax_kind, + type_checking_ranges, + conditional_import_ranges, + ) for module_name in _extract_dynamic_import_names(node): - imports.append(_LocatedImport(module_name, kind, node.lineno)) + imports.append( + _LocatedImport(module_name, kind, node.lineno, syntax_kind) + ) return [ import_ @@ -389,6 +415,43 @@ def _extract_dynamic_import_names(node: ast.Call) -> list[str]: return [] +def _import_from_module_names(node: ast.ImportFrom) -> tuple[str, ...]: + """Return resolvable module names for a from-import statement.""" + dots = "." * (node.level or 0) + if node.module: + return (f"{dots}{node.module}",) + + aliases = tuple(alias.name for alias in node.names if alias.name != "*") + if dots and aliases: + return tuple(f"{dots}{alias}" for alias in aliases) + return (dots,) if dots else () + + +def _edge_import_kinds(import_: _LocatedImport) -> tuple[ImportKind, ...]: + """Return graph labels without losing syntax for conditional imports.""" + resolution_kind = import_.resolution_kind or import_.import_kind + if ( + import_.import_kind == ImportKind.CONDITIONAL_IMPORT + and resolution_kind != import_.import_kind + ): + return (resolution_kind, import_.import_kind) + return (import_.import_kind,) + + +def _classify_import( + node: ast.AST, + default_kind: ImportKind, + type_checking_ranges: list[tuple[int, int]], + conditional_import_ranges: list[tuple[int, int]], +) -> ImportKind: + """Classify an import node by special context before syntax kind.""" + if _in_type_checking(node, type_checking_ranges): + return ImportKind.TYPE_IMPORT + if _in_conditional_import(node, conditional_import_ranges): + return ImportKind.CONDITIONAL_IMPORT + return default_kind + + def _find_type_checking_ranges(tree: ast.Module) -> list[tuple[int, int]]: """Find line ranges of TYPE_CHECKING blocks.""" ranges: list[tuple[int, int]] = [] @@ -414,6 +477,46 @@ def _find_type_checking_ranges(tree: ast.Module) -> list[tuple[int, int]]: return sorted(ranges, key=lambda ele: ele[0]) +def _find_conditional_import_ranges(tree: ast.Module) -> list[tuple[int, int]]: + """Find try/except ImportError ranges that contain optional imports.""" + ranges: list[tuple[int, int]] = [] + + for node in ast.walk(tree): + if not isinstance(node, ast.Try): + continue + if not any(_handles_import_error(handler.type) for handler in node.handlers): + continue + + ranges.extend(_statement_ranges(node.body)) + for handler in node.handlers: + if _handles_import_error(handler.type): + ranges.extend(_statement_ranges(handler.body)) + + return sorted(ranges, key=lambda ele: ele[0]) + + +def _handles_import_error(node: ast.expr | None) -> bool: + """Return True if an except handler catches import-related errors.""" + if node is None: + return False + if isinstance(node, ast.Name): + return node.id in {"ImportError", "ModuleNotFoundError"} + if isinstance(node, ast.Attribute): + return node.attr in {"ImportError", "ModuleNotFoundError"} + if isinstance(node, ast.Tuple): + return any(_handles_import_error(elt) for elt in node.elts) + return False + + +def _statement_ranges(statements: list[ast.stmt]) -> list[tuple[int, int]]: + """Return line ranges covered by statement blocks.""" + if not statements: + return [] + start = statements[0].lineno + end = max(getattr(statement, "end_lineno", statement.lineno) for statement in statements) + return [(start, end)] + + def _in_type_checking(node: ast.AST, ranges: list[tuple[int, int]]) -> bool: """Check if a node is inside a TYPE_CHECKING block.""" if not hasattr(node, "lineno"): @@ -422,6 +525,14 @@ def _in_type_checking(node: ast.AST, ranges: list[tuple[int, int]]) -> bool: return any(start <= lineno <= end for start, end in ranges) +def _in_conditional_import(node: ast.AST, ranges: list[tuple[int, int]]) -> bool: + """Check if a node is inside a try/except ImportError block.""" + if not hasattr(node, "lineno"): + return False + lineno = node.lineno + return any(start <= lineno <= end for start, end in ranges) + + def _resolve_import( import_name: str, source_file: str, @@ -433,7 +544,15 @@ def _resolve_import( Returns (resolved_path, is_external). The path is normalized with forward slashes. """ - if kind in (ImportKind.RELATIVE_IMPORT, ImportKind.TYPE_IMPORT) and import_name.startswith("."): + if ( + kind + in ( + ImportKind.RELATIVE_IMPORT, + ImportKind.TYPE_IMPORT, + ImportKind.CONDITIONAL_IMPORT, + ) + and import_name.startswith(".") + ): # Relative import return _resolve_relative_import(import_name, source_file, project_root) diff --git a/src/archunitpython/common/extraction/graph.py b/src/archunitpython/common/extraction/graph.py index 79e7a61..5ace535 100644 --- a/src/archunitpython/common/extraction/graph.py +++ b/src/archunitpython/common/extraction/graph.py @@ -14,6 +14,7 @@ class ImportKind(Enum): RELATIVE_IMPORT = "relative" # from . import bar / from ..foo import bar DYNAMIC_IMPORT = "dynamic" # __import__('foo') / importlib.import_module() TYPE_IMPORT = "type" # inside TYPE_CHECKING block + CONDITIONAL_IMPORT = "conditional" # inside try/except ImportError @dataclass(frozen=True) diff --git a/tests/common/test_extract_graph.py b/tests/common/test_extract_graph.py index 6372e5e..37c0d0d 100644 --- a/tests/common/test_extract_graph.py +++ b/tests/common/test_extract_graph.py @@ -106,6 +106,32 @@ def test_importlib_import_module(self): finally: shutil.rmtree(project_root, ignore_errors=True) + def test_conditional_import(self): + temp_root = Path(__file__).resolve().parent / ".tmp" + temp_root.mkdir(exist_ok=True) + project_root = temp_root / f"project_{uuid4().hex}" + project_root.mkdir() + file_path = project_root / "loader.py" + file_path.write_text( + "\n".join( + [ + "try:", + " import orjson", + "except ImportError:", + " import json", + "", + ] + ), + encoding="utf-8", + ) + + try: + imports = _extract_imports(str(file_path)) + assert ("orjson", ImportKind.CONDITIONAL_IMPORT) in imports + assert ("json", ImportKind.CONDITIONAL_IMPORT) in imports + finally: + shutil.rmtree(project_root, ignore_errors=True) + class TestExtractGraph: def setup_method(self): @@ -378,6 +404,309 @@ def test_dynamic_import_resolves_to_internal_edge(self): assert ImportKind.DYNAMIC_IMPORT in edges[0].import_kinds +class TestConditionalImportGraphHandling: + def setup_method(self): + clear_graph_cache() + + def _build_conditional_project( + self, + service_source: str | None = None, + *, + service_subdirectory: str | None = None, + ) -> str: + temp_root = Path(__file__).resolve().parent / ".tmp" + temp_root.mkdir(exist_ok=True) + project_root = temp_root / f"project_{uuid4().hex}" + project_root.mkdir() + + package_dir = project_root / "sample_project" + package_dir.mkdir(parents=True, exist_ok=True) + + (package_dir / "__init__.py").write_text("", encoding="utf-8") + (package_dir / "fast_model.py").write_text( + "class FastUser:\n pass\n", + encoding="utf-8", + ) + (package_dir / "fallback_model.py").write_text( + "class FallbackUser:\n pass\n", + encoding="utf-8", + ) + if service_source is None: + service_source = "\n".join( + [ + "try:", + " from sample_project.fast_model import FastUser", + "except ImportError:", + " from sample_project.fallback_model import FallbackUser", + "", + ] + ) + + service_dir = package_dir + if service_subdirectory is not None: + service_dir = package_dir / service_subdirectory + service_dir.mkdir() + (service_dir / "__init__.py").write_text("", encoding="utf-8") + (service_dir / "service.py").write_text(service_source, encoding="utf-8") + self._service_subdirectory = service_subdirectory + 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 test_import_error_fallback_imports_are_marked_conditional(self): + project_root = self._build_conditional_project() + + graph = extract_graph(project_root) + service_path = os.path.abspath( + os.path.join(project_root, "sample_project", "service.py") + ).replace("\\", "/") + target_paths = { + os.path.abspath( + os.path.join(project_root, "sample_project", "fast_model.py") + ).replace("\\", "/"), + os.path.abspath( + os.path.join(project_root, "sample_project", "fallback_model.py") + ).replace("\\", "/"), + } + + edges = [ + edge + for edge in graph + if edge.source == service_path and edge.target in target_paths + ] + + assert len(edges) == 2 + assert all(edge.external is False for edge in edges) + assert all(ImportKind.CONDITIONAL_IMPORT in edge.import_kinds for edge in edges) + assert all(ImportKind.FROM_IMPORT in edge.import_kinds for edge in edges) + + def _conditional_edges(self, project_root: str) -> list[Edge]: + service_parts = [project_root, "sample_project"] + service_subdirectory = getattr(self, "_service_subdirectory", None) + if service_subdirectory is not None: + service_parts.append(service_subdirectory) + service_parts.append("service.py") + service_path = os.path.abspath(os.path.join(*service_parts)).replace("\\", "/") + return [ + edge + for edge in extract_graph(project_root) + if edge.source == service_path and edge.target != service_path + ] + + def test_relative_fallback_imports_resolve_sibling_modules(self): + project_root = self._build_conditional_project( + "\n".join( + [ + "try:", + " from . import fast_model", + "except ImportError:", + " from . import fallback_model", + "", + ] + ) + ) + + edges = self._conditional_edges(project_root) + target_paths = { + os.path.abspath( + os.path.join(project_root, "sample_project", module) + ).replace("\\", "/") + for module in ("fast_model.py", "fallback_model.py") + } + + assert {edge.target for edge in edges if not edge.external} == target_paths + assert all(ImportKind.RELATIVE_IMPORT in edge.import_kinds for edge in edges) + assert all(ImportKind.CONDITIONAL_IMPORT in edge.import_kinds for edge in edges) + assert not any(edge.target.endswith("/__init__.py") for edge in edges) + + def test_conditional_relative_import_resolves_multiple_aliased_modules(self): + project_root = self._build_conditional_project( + "\n".join( + [ + "try:", + " from . import fast_model as selected, fallback_model", + "except ImportError:", + " pass", + "", + ] + ) + ) + + edges = self._conditional_edges(project_root) + target_paths = { + os.path.abspath( + os.path.join(project_root, "sample_project", module) + ).replace("\\", "/") + for module in ("fast_model.py", "fallback_model.py") + } + + assert {edge.target for edge in edges if not edge.external} == target_paths + assert all(ImportKind.RELATIVE_IMPORT in edge.import_kinds for edge in edges) + assert all(ImportKind.CONDITIONAL_IMPORT in edge.import_kinds for edge in edges) + + def test_regular_relative_import_resolves_without_conditional_kind(self): + project_root = self._build_conditional_project("from . import fast_model\n") + + model_edges = [ + edge + for edge in self._conditional_edges(project_root) + if edge.target.endswith("/fast_model.py") + ] + + assert len(model_edges) == 1 + assert ImportKind.RELATIVE_IMPORT in model_edges[0].import_kinds + assert ImportKind.CONDITIONAL_IMPORT not in model_edges[0].import_kinds + + def test_parent_relative_fallback_imports_resolve_modules(self): + project_root = self._build_conditional_project( + "\n".join( + [ + "try:", + " from .. import fast_model", + "except ImportError:", + " from .. import fallback_model", + "", + ] + ), + service_subdirectory="feature", + ) + + edges = self._conditional_edges(project_root) + target_paths = { + os.path.abspath( + os.path.join(project_root, "sample_project", module) + ).replace("\\", "/") + for module in ("fast_model.py", "fallback_model.py") + } + + assert {edge.target for edge in edges if not edge.external} == target_paths + assert all(ImportKind.RELATIVE_IMPORT in edge.import_kinds for edge in edges) + assert all(ImportKind.CONDITIONAL_IMPORT in edge.import_kinds for edge in edges) + + def test_module_not_found_error_marks_fallback_conditional(self): + project_root = self._build_conditional_project( + "\n".join( + [ + "try:", + " from . import fast_model", + "except ModuleNotFoundError:", + " from . import fallback_model", + "", + ] + ) + ) + + edges = self._conditional_edges(project_root) + + assert len(edges) == 2 + assert all(ImportKind.CONDITIONAL_IMPORT in edge.import_kinds for edge in edges) + + def test_tuple_handler_marks_import_fallback_conditional(self): + project_root = self._build_conditional_project( + "\n".join( + [ + "try:", + " from . import fast_model", + "except (ImportError, OSError):", + " from . import fallback_model", + "", + ] + ) + ) + + edges = self._conditional_edges(project_root) + + assert len(edges) == 2 + assert all(ImportKind.CONDITIONAL_IMPORT in edge.import_kinds for edge in edges) + + def test_non_import_error_handler_does_not_mark_imports_conditional(self): + project_root = self._build_conditional_project( + "\n".join( + [ + "try:", + " from sample_project.fast_model import FastUser", + "except OSError:", + " from sample_project.fallback_model import FallbackUser", + "", + ] + ) + ) + + edges = self._conditional_edges(project_root) + + assert len(edges) == 2 + assert all(ImportKind.FROM_IMPORT in edge.import_kinds for edge in edges) + assert all( + ImportKind.CONDITIONAL_IMPORT not in edge.import_kinds for edge in edges + ) + + def test_conditional_dynamic_import_retains_dynamic_kind(self): + project_root = self._build_conditional_project( + "\n".join( + [ + "import importlib", + "", + "try:", + ' importlib.import_module("sample_project.fast_model")', + "except ImportError:", + ' importlib.import_module("sample_project.fallback_model")', + "", + ] + ) + ) + + internal_edges = [ + edge for edge in self._conditional_edges(project_root) if not edge.external + ] + + assert len(internal_edges) == 2 + assert all( + ImportKind.CONDITIONAL_IMPORT in edge.import_kinds + for edge in internal_edges + ) + assert all(ImportKind.DYNAMIC_IMPORT in edge.import_kinds for edge in internal_edges) + + def test_type_checking_import_takes_precedence_over_conditional_context(self): + project_root = self._build_conditional_project( + "\n".join( + [ + "from typing import TYPE_CHECKING", + "", + "try:", + " if TYPE_CHECKING:", + " from . import fast_model", + "except ImportError:", + " pass", + "", + ] + ) + ) + + edges = self._conditional_edges(project_root) + model_edges = [edge for edge in edges if edge.target.endswith("/fast_model.py")] + + assert len(model_edges) == 1 + assert ImportKind.TYPE_IMPORT in model_edges[0].import_kinds + assert ImportKind.CONDITIONAL_IMPORT not in model_edges[0].import_kinds + + ignored_graph = extract_graph( + project_root, + options=CheckOptions( + clear_cache=True, + ignore_type_checking_imports=True, + ), + ) + assert not any( + edge.source.endswith("/service.py") + and edge.target.endswith("/fast_model.py") + for edge in ignored_graph + ) + + class TestIgnoreDirectives: def setup_method(self): clear_graph_cache() @@ -469,3 +798,4 @@ def test_all_kinds_exist(self): assert ImportKind.RELATIVE_IMPORT.value == "relative" assert ImportKind.DYNAMIC_IMPORT.value == "dynamic" assert ImportKind.TYPE_IMPORT.value == "type" + assert ImportKind.CONDITIONAL_IMPORT.value == "conditional"