Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion BACKLOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
9 changes: 9 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
159 changes: 139 additions & 20 deletions src/archunitpython/common/extraction/extract_graph.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@ class _LocatedImport:
module_name: str
import_kind: ImportKind
line_number: int
resolution_kind: ImportKind | None = None


@dataclass(frozen=True)
Expand Down Expand Up @@ -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
Expand All @@ -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),
)
)

Expand Down Expand Up @@ -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_
Expand Down Expand Up @@ -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]] = []
Expand All @@ -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"):
Expand All @@ -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,
Expand All @@ -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)

Expand Down
1 change: 1 addition & 0 deletions src/archunitpython/common/extraction/graph.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
Loading