diff --git a/kanta/__main__.py b/kanta/__main__.py index b461b87..8a2e501 100644 --- a/kanta/__main__.py +++ b/kanta/__main__.py @@ -54,7 +54,9 @@ class _CliError(Exception): def _import_dotted(path: str) -> Any: - """Import ``module.submodule.Attr`` and return the attribute.""" + """Import ``module.submodule.Attr`` or a filesystem path and return the attribute.""" + if _is_file_path(path): + return _import_from_file(path) if "." not in path: raise ValueError(f"dotted path must contain a dot: {path!r}") module_name, attr_name = path.rsplit(".", 1) @@ -65,13 +67,71 @@ def _import_dotted(path: str) -> Any: raise ImportError(f"{path!r} not found in {module_name!r}") from exc +def _is_file_path(path: str) -> bool: + """Return True if *path* looks like a filesystem path rather than a dotted name.""" + return "/" in path or "\\" in path or ":" in path + + +def _import_from_file(path: str) -> Any: + """Import a module or attribute from a filesystem path. + + *path* may be ``path/to/file.py`` (returns the module) or + ``path/to/file.py:symbol`` (returns ``symbol`` from the module). + """ + if ":" in path: + file_path, symbol = path.rsplit(":", 1) + else: + file_path, symbol = path, None + + file_path = Path(file_path).resolve() + if not file_path.exists(): + raise ImportError(f"{file_path!r} not found") + if not file_path.is_file(): + raise ImportError(f"{file_path!r} is not a file") + + module_name = f"_kanta_cli_{file_path.stem}_{file_path.stat().st_ino}" + spec = importlib.util.spec_from_file_location(module_name, file_path) + if spec is None or spec.loader is None: + raise ImportError(f"cannot load {file_path!r}") + module = importlib.util.module_from_spec(spec) + + file_dir = str(file_path.parent) + added_dir = False + if file_dir not in sys.path: + sys.path.insert(0, file_dir) + added_dir = True + try: + sys.modules[module_name] = module + spec.loader.exec_module(module) + finally: + if added_dir: + sys.path.remove(file_dir) + + if symbol is None: + return module + try: + return getattr(module, symbol) + except AttributeError as exc: + raise ImportError(f"{symbol!r} not found in {file_path!r}") from exc + + def _import_kanta_object(path: str) -> Any: - """Import a Kanta object by module path. + """Import a Kanta object by module or filesystem path. If ``path`` names an importable module, look up an object named - ``kanta`` in it; otherwise treat ``path`` as ``module.attr`` referring - directly to the object. + ``kanta`` in it; otherwise treat ``path`` as ``module.attr`` or + ``path/to/file.py[:kanta]`` referring directly to the object. """ + if _is_file_path(path): + if ":" in path: + return _import_from_file(path) + module = _import_from_file(path) + try: + return getattr(module, "kanta") + except AttributeError as exc: + raise ImportError( + f"no 'kanta' object found in {path!r}" + ) from exc try: spec = importlib.util.find_spec(path) except ImportError: @@ -105,24 +165,30 @@ def _parse_args(argv: list[str] | None = None) -> argparse.Namespace: "-d", "--data", metavar="MOD", - help="Dotted path to the root data type (e.g. myapp.models.Data).", + help=( + "Dotted path or filesystem path to the root data type." + " Examples: myapp.models.Data, myapp/models.py:Data." + ), ) parser.add_argument( "-m", "--migrations", metavar="MOD", - help="Dotted path to the migrations module (e.g. myapp.migrations).", + help=( + "Dotted path or filesystem path to the migrations module." + " Examples: myapp.migrations, myapp/migrations.py." + ), ) parser.add_argument( "-k", "--kanta", metavar="MOD", help=( - "Module path to an existing Kanta object to use: either a module" - " containing an object named 'kanta' (e.g. myapp.db) or a dotted" - " path to the object itself (e.g. myapp.db.kanta). Its type," - " migrations, and logfmt/logemit callbacks are used. Cannot be" - " combined with -d or -m." + "Module path or filesystem path to an existing Kanta object to use." + " Either a module containing an object named 'kanta' (e.g. myapp.db)," + " a dotted path to the object (e.g. myapp.db.kanta), or a file path" + " (e.g. myapp/db.py or myapp/db.py:kanta). Its type, migrations, and" + " logfmt/logemit callbacks are used. Cannot be combined with -d or -m." ), ) parser.add_argument( diff --git a/tests/test_cli.py b/tests/test_cli.py index 405f182..b28e3d6 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -3,7 +3,13 @@ import sys from datetime import UTC, datetime -from kanta.__main__ import _extra_import_paths, _format_ts, main +from kanta.__main__ import ( + _extra_import_paths, + _format_ts, + _import_dotted, + _import_kanta_object, + main, +) from kanta.serialization import JsonSerializer from kanta.serialization.framing import LineFramer from kanta.structs import ChangeRecord, Snapshot @@ -98,3 +104,27 @@ def test_cli_snapshot_line_format(tmp_path, capsys): assert "\x1b[97msnapshot s0" in err assert "\x1b[38;5;250m v1 2026-08-12 09:00:00" in err assert "\x1b[38;5;242m 13 B" in err + + +def test_import_dotted_from_file_path(tmp_path): + """--data can be a filesystem path with an optional colon-separated symbol.""" + module = tmp_path / "models.py" + module.write_text("class Data:\n pass\n") + result = _import_dotted(f"{module}:Data") + assert result.__name__ == "Data" + + +def test_import_kanta_object_from_file_path(tmp_path): + """--kanta can be a filesystem path; default symbol is ``kanta``.""" + module = tmp_path / "database.py" + module.write_text("class Kanta:\n pass\nkanta = Kanta()\n") + result = _import_kanta_object(str(module)) + assert type(result).__name__ == "Kanta" + + +def test_import_kanta_object_from_file_path_with_symbol(tmp_path): + """--kanta can be a filesystem path with an explicit colon-separated symbol.""" + module = tmp_path / "database.py" + module.write_text("class CustomKanta:\n pass\nmy_kanta = CustomKanta()\n") + result = _import_kanta_object(f"{module}:my_kanta") + assert type(result).__name__ == "CustomKanta"