Skip to content
4 changes: 2 additions & 2 deletions python/deptry/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@
from deptry.dependency_getter.requirements_txt import RequirementsTxtDependencyGetter
from deptry.dependency_specification_detector import DependencyManagementFormat, DependencySpecificationDetector
from deptry.exceptions import IncorrectDependencyFormatError, UnsupportedPythonVersionError
from deptry.imports.extract import get_imported_modules_from_list_of_files
from deptry.imports.extract import ImportExtractor
from deptry.module import ModuleBuilder, ModuleLocations
from deptry.python_file_finder import PythonFileFinder
from deptry.reporters import JSONReporter, TextReporter
Expand Down Expand Up @@ -80,7 +80,7 @@ def run(self) -> None:
).build(),
locations,
)
for module, locations in get_imported_modules_from_list_of_files(all_python_files).items()
for module, locations in ImportExtractor().get_imported_modules_from_list_of_files(all_python_files).items()
]
imported_modules_with_locations = [
module_with_locations
Expand Down
76 changes: 43 additions & 33 deletions python/deptry/imports/extract.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,8 @@
from __future__ import annotations

import json
import logging
from collections import defaultdict
from collections import OrderedDict, defaultdict
from typing import TYPE_CHECKING

from deptry.rust import get_imports_from_ipynb_files, get_imports_from_py_files
Expand All @@ -14,35 +15,44 @@
from deptry.imports.location import Location


def get_imported_modules_from_list_of_files(list_of_files: list[Path]) -> dict[str, list[Location]]:
logging.info("Scanning %d %s...", len(list_of_files), "files" if len(list_of_files) > 1 else "file")

py_files = [str(file) for file in list_of_files if file.suffix == ".py"]
ipynb_files = [str(file) for file in list_of_files if file.suffix == ".ipynb"]

modules: dict[str, list[Location]] = defaultdict(list)

# Process all .py files in parallel using Rust
if py_files:
rust_result = get_imports_from_py_files(py_files)
for module, locations in convert_rust_locations_to_python_locations(rust_result).items():
modules[module].extend(locations)

# Process all .ipynb files in parallel using Rust
if ipynb_files:
rust_result = get_imports_from_ipynb_files(ipynb_files)
for module, locations in convert_rust_locations_to_python_locations(rust_result).items():
modules[module].extend(locations)

logging.debug("All imported modules: %s\n", modules)

return modules


def convert_rust_locations_to_python_locations(
imported_modules: dict[str, list[RustLocation]],
) -> dict[str, list[Location]]:
converted_modules: dict[str, list[Location]] = {}
for module, locations in imported_modules.items():
converted_modules[module] = [Location.from_rust_location_object(loc) for loc in locations]
return converted_modules
class ImportExtractor:
def get_imported_modules_from_list_of_files(self, list_of_files: list[Path]) -> dict[str, list[Location]]:
logging.info("Scanning %d %s...", len(list_of_files), "files" if len(list_of_files) > 1 else "file")

py_files = [str(file) for file in list_of_files if file.suffix == ".py"]
ipynb_files = [str(file) for file in list_of_files if file.suffix == ".ipynb"]

modules: dict[str, list[Location]] = defaultdict(list)

# Process all .py files in parallel using Rust
if py_files:
rust_result = get_imports_from_py_files(py_files)
for module, locations in self._convert_rust_locations_to_python_locations(rust_result).items():
modules[module].extend(locations)

# Process all .ipynb files in parallel using Rust
if ipynb_files:
rust_result = get_imports_from_ipynb_files(ipynb_files)
for module, locations in self._convert_rust_locations_to_python_locations(rust_result).items():
modules[module].extend(locations)

sorted_modules = OrderedDict(sorted(modules.items()))
self._log_modules_with_locations(sorted_modules)
return sorted_modules

@staticmethod
def _log_modules_with_locations(modules: dict[str, list[Location]]) -> None:
modules_dict = {
module_name: [str(location) for location in locations] for module_name, locations in modules.items()
}
modules_json = json.dumps(modules_dict, indent=2)
logging.debug("All imported modules and their locations:\n%s", modules_json)

@staticmethod
def _convert_rust_locations_to_python_locations(
imported_modules: dict[str, list[RustLocation]],
) -> dict[str, list[Location]]:
converted_modules: dict[str, list[Location]] = {}
for module, locations in imported_modules.items():
converted_modules[module] = [Location.from_rust_location_object(loc) for loc in locations]
return converted_modules
22 changes: 14 additions & 8 deletions tests/unit/imports/test_extract.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@

import pytest

from deptry.imports.extract import get_imported_modules_from_list_of_files
from deptry.imports.extract import ImportExtractor
from deptry.imports.location import Location
from tests.utils import run_within_dir

Expand All @@ -20,7 +20,7 @@
def test_import_parser_py() -> None:
some_imports_path = Path("tests/data/some_imports.py")

assert get_imported_modules_from_list_of_files([some_imports_path]) == {
assert ImportExtractor().get_imported_modules_from_list_of_files([some_imports_path]) == {
"barfoo": [Location(some_imports_path, 20, 8)],
"baz": [Location(some_imports_path, 16, 5)],
"click": [Location(some_imports_path, 24, 12)],
Expand All @@ -44,7 +44,7 @@ def test_import_parser_py() -> None:
def test_import_parser_ipynb() -> None:
notebook_path = Path("tests/data/example_project/src/notebook.ipynb")

assert get_imported_modules_from_list_of_files([notebook_path]) == {
assert ImportExtractor().get_imported_modules_from_list_of_files([notebook_path]) == {
"click": [Location(notebook_path, 1, 8)],
"toml": [Location(notebook_path, 5, 8)],
"urllib3": [Location(notebook_path, 3, 1)],
Expand Down Expand Up @@ -79,7 +79,9 @@ def test_import_parser_file_encodings(file_content: str, encoding: str | None, t
with random_file.open("w", encoding=encoding) as f:
f.write(file_content)

assert get_imported_modules_from_list_of_files([random_file]) == {"foo": [Location(random_file, 2, 8)]}
assert ImportExtractor().get_imported_modules_from_list_of_files([random_file]) == {
"foo": [Location(random_file, 2, 8)]
}


@pytest.mark.parametrize(
Expand Down Expand Up @@ -119,7 +121,9 @@ def test_import_parser_file_encodings_ipynb(code_cell_content: list[str], encodi
}
f.write(json.dumps(file_content))

assert get_imported_modules_from_list_of_files([random_file]) == {"foo": [Location(random_file, 1, 8)]}
assert ImportExtractor().get_imported_modules_from_list_of_files([random_file]) == {
"foo": [Location(random_file, 1, 8)]
}


def test_import_parser_errors(tmp_path: Path, caplog: LogCaptureFixture) -> None:
Expand All @@ -138,7 +142,7 @@ def test_import_parser_errors(tmp_path: Path, caplog: LogCaptureFixture) -> None
f.write("invalid_syntax:::")

with caplog.at_level(logging.WARNING):
assert get_imported_modules_from_list_of_files([
assert ImportExtractor().get_imported_modules_from_list_of_files([
file_ok,
file_with_bad_encoding,
file_with_syntax_error,
Expand Down Expand Up @@ -185,7 +189,7 @@ def test_import_parser_for_ipynb_errors(tmp_path: Path, caplog: LogCaptureFixtur

# Execute function and assert the result for well-formed notebook
with caplog.at_level(logging.WARNING):
assert get_imported_modules_from_list_of_files([
assert ImportExtractor().get_imported_modules_from_list_of_files([
notebook_ok,
notebook_with_syntax_error,
]) == {"numpy": [Location(file=Path("notebook_ok.ipynb"), line=1, column=8)]}
Expand All @@ -203,4 +207,6 @@ def test_python_3_12_f_string_syntax(tmp_path: Path) -> None:
with file_path.open("w") as f:
f.write('import foo\nprint(f"abc{"def"}")')

assert get_imported_modules_from_list_of_files([file_path]) == {"foo": [Location(file_path, 1, 8)]}
assert ImportExtractor().get_imported_modules_from_list_of_files([file_path]) == {
"foo": [Location(file_path, 1, 8)]
}