Scope sys.path additions around dynamic imports and include nearby .venv site-packages.

kanta/__main__.py

@@ -4,6 +4,7 @@ from __future__ import annotations
 
 import argparse
 import asyncio
+import contextlib
 import importlib
 import importlib.util
 import logging
@@ -196,6 +197,46 @@ def _format_size(n: int) -> str:
     return f"{n / (1024 * 1024):.1f} MB"
 
 
+def _find_venv_site_packages(start: Path) -> list[Path]:
+    """Return site-packages dirs of ``.venv`` directories from *start* to parents."""
+    found: list[Path] = []
+    for parent in [start, *start.parents]:
+        venv = parent / ".venv"
+        if not venv.is_dir():
+            continue
+        for site_packages in venv.glob("lib/python*/site-packages"):
+            found.append(site_packages)
+            break
+        else:
+            win_site = venv / "Lib" / "site-packages"
+            if win_site.is_dir():
+                found.append(win_site)
+    return found
+
+
+@contextlib.contextmanager
+def _extra_import_paths():
+    """Temporarily add current dir and nearby venv site-packages to ``sys.path``.
+
+    The current directory is inserted first, then local ``.venv`` site-packages,
+    then any parent ``.venv`` site-packages.  Only paths that were not already
+    present are added, and only those added paths are removed on exit.
+    """
+    paths_to_add = [str(Path.cwd())]
+    paths_to_add.extend(str(p) for p in _find_venv_site_packages(Path.cwd()))
+    added: list[str] = []
+    for path in reversed(paths_to_add):
+        if path not in sys.path:
+            sys.path.insert(0, path)
+            added.append(path)
+    try:
+        yield
+    finally:
+        for path in added:
+            if path in sys.path:
+                sys.path.remove(path)
+
+
 def _print_snapshot_indicator(
     label: str,
     snap: Snapshot,
@@ -305,6 +346,7 @@ async def _run(args: argparse.Namespace) -> int:
 
     data_type: type[Any] | None = None
     if args.data:
+        with _extra_import_paths():
             try:
                 data_type = _import_dotted(args.data)
             except (ImportError, ValueError) as exc:
@@ -314,6 +356,7 @@ async def _run(args: argparse.Namespace) -> int:
     kanta_owned = False
     kanta_typed: Kanta[Any] | None = None
     try:
+        with _extra_import_paths():
             kanta, kanta_owned = _get_kanta(args, filename)
         if data_type is None and args.kanta and kanta._impl.data_type is not dict:
             data_type = kanta._impl.data_type
@@ -478,8 +521,6 @@ async def _run(args: argparse.Namespace) -> int:
 
 def main(argv: list[str] | None = None) -> int:
     """Entry point for ``python -m kanta``."""
-    if "." not in sys.path:
-        sys.path.insert(0, ".")
     args = _parse_args(argv)
     try:
         return asyncio.run(_run(args))

tests/test_cli.py

@@ -1,8 +1,9 @@
 """Tests for the ``python -m kanta`` CLI output formatting."""
 
+import sys
 from datetime import UTC, datetime
 
-from kanta.__main__ import _format_ts, main
+from kanta.__main__ import _extra_import_paths, _format_ts, main
 from kanta.serialization import JsonSerializer
 from kanta.serialization.framing import LineFramer
 from kanta.structs import ChangeRecord, Snapshot
@@ -14,6 +15,42 @@ def test_format_ts_strips_microseconds():
     assert _format_ts(dt) == "2026-08-12 10:06:52"
 
 
+def test_extra_import_paths_are_temporary(tmp_path, monkeypatch):
+    """CWD and nearby venv site-packages are added only for the import block."""
+    parent_dir = tmp_path / "parent"
+    cwd = parent_dir / "child"
+    venv_site = (
+        cwd
+        / ".venv"
+        / "lib"
+        / f"python{sys.version_info.major}.{sys.version_info.minor}"
+        / "site-packages"
+    )
+    venv_site.mkdir(parents=True)
+    parent_venv_site = (
+        parent_dir
+        / ".venv"
+        / "lib"
+        / f"python{sys.version_info.major}.{sys.version_info.minor}"
+        / "site-packages"
+    )
+    parent_venv_site.mkdir(parents=True)
+
+    monkeypatch.chdir(cwd)
+    cwd_str = str(cwd)
+    venv = str(venv_site)
+    parent_venv = str(parent_venv_site)
+
+    before = sys.path.copy()
+    with _extra_import_paths():
+        during = sys.path.copy()
+        assert cwd_str in during
+        assert venv in during
+        assert parent_venv in during
+        assert during.index(cwd_str) < during.index(venv) < during.index(parent_venv)
+    assert sys.path == before
+
+
 def test_cli_snapshot_line_format(tmp_path, capsys):
     """Snapshot lines are timestamped and colored with metadata."""
     path = tmp_path / "test.kantadb"