From dd1c82fe4d43e7809ba463fb6ba7a77e3a9d3aa4 Mon Sep 17 00:00:00 2001 From: Ramesh Padmanabhaiah Date: Sun, 5 Jul 2026 18:39:37 -0700 Subject: [PATCH] refactor: split base cli runtime helpers --- lib/python/base_cli/_runtime.py | 68 ++++++++++++++++++ lib/python/base_cli/app.py | 72 ++++--------------- .../tests/test_app_runtime_boundary.py | 32 +++++++++ 3 files changed, 112 insertions(+), 60 deletions(-) create mode 100644 lib/python/base_cli/_runtime.py create mode 100644 lib/python/base_cli/tests/test_app_runtime_boundary.py diff --git a/lib/python/base_cli/_runtime.py b/lib/python/base_cli/_runtime.py new file mode 100644 index 00000000..41a7869b --- /dev/null +++ b/lib/python/base_cli/_runtime.py @@ -0,0 +1,68 @@ +from __future__ import annotations + +import logging +from dataclasses import dataclass +from pathlib import Path + + +@dataclass(frozen=True) +class RuntimeLayout: + state_dir: Path + log_dir: Path + cache_dir: Path + temp_dir: Path + + +def runtime_layout(cache_root: Path, cli_name: str, run_id: str) -> RuntimeLayout: + state_dir = cache_root / "cli" / cli_name + return RuntimeLayout( + state_dir=state_dir, + log_dir=state_dir / "logs", + cache_dir=state_dir / "cache", + temp_dir=state_dir / "tmp" / run_id, + ) + + +def create_runtime_directory(path: Path, cache_root: Path) -> None: + try: + path.mkdir(parents=True, exist_ok=True) + except OSError as exc: + raise RuntimeError(_runtime_directory_error(path, cache_root, exc)) from exc + + +def prune_log_files( + log_dir: Path, + current_log_file: Path, + max_log_files: int, + logger: logging.Logger, +) -> None: + candidates: list[tuple[str, Path]] = [] + for path in log_dir.glob("*.log"): + if _same_path(path, current_log_file): + continue + candidates.append((path.name, path)) + + excess_count = len(candidates) + 1 - max_log_files + if excess_count <= 0: + return + + for _, path in sorted(candidates)[:excess_count]: + try: + path.unlink() + except OSError as exc: + logger.warning("Could not prune log file '%s': %s", path, exc) + + +def _runtime_directory_error(path: Path, cache_root: Path, exc: OSError) -> str: + return ( + f"Unable to create Base runtime directory '{path}': {exc}. " + f"Check permissions on that directory. If the Base cache root '{cache_root}' is unusable, " + "set BASE_CACHE_DIR to a writable directory." + ) + + +def _same_path(left: Path, right: Path) -> bool: + try: + return left.resolve() == right.resolve() + except OSError: + return left.absolute() == right.absolute() diff --git a/lib/python/base_cli/app.py b/lib/python/base_cli/app.py index bcb82fd3..bace0068 100644 --- a/lib/python/base_cli/app.py +++ b/lib/python/base_cli/app.py @@ -1,13 +1,13 @@ from __future__ import annotations import functools -import logging import os import sys from contextvars import ContextVar from pathlib import Path from typing import Any, Callable +from ._runtime import create_runtime_directory, prune_log_files, runtime_layout from .config import load_config, read_user_config from .context import Context, reset_current_context, set_current_context from .exit_codes import ExitCode @@ -178,26 +178,23 @@ def _create_context(self, standard: dict[str, Any], sensitive_options: set[str], keep_temp = bool(standard.get("keep_temp") or config.get("keep_temp")) cache_root = base_cache_root() - state_dir = cache_root / "cli" / self.name - log_dir = state_dir / "logs" - cache_dir = state_dir / "cache" - temp_dir = state_dir / "tmp" / run_id + layout = runtime_layout(cache_root, self.name, run_id) log_file = Path(standard["log_file"]).expanduser() if standard.get("log_file") else None uses_default_log_file = log_file is None if dry_run or not self.log_to_file: if log_file is not None: - _create_runtime_directory(log_file.parent, cache_root) + create_runtime_directory(log_file.parent, cache_root) else: - for directory in (log_dir, cache_dir, temp_dir): - _create_runtime_directory(directory, cache_root) + for directory in (layout.log_dir, layout.cache_dir, layout.temp_dir): + create_runtime_directory(directory, cache_root) if log_file is None: - log_file = log_dir / f"{run_id}.log" - _create_runtime_directory(log_file.parent, cache_root) + log_file = layout.log_dir / f"{run_id}.log" + create_runtime_directory(log_file.parent, cache_root) logger = configure_logger(self.name, log_file, debug, quiet=quiet) logger.debug("cli=%s run_id=%s environment=%s", self.name, run_id, environment) if self.max_log_files is not None and uses_default_log_file and log_file is not None: - _prune_log_files(log_dir, log_file, self.max_log_files, logger) + prune_log_files(layout.log_dir, log_file, self.max_log_files, logger) return Context( cli_name=self.name, @@ -206,10 +203,10 @@ def _create_context(self, standard: dict[str, Any], sensitive_options: set[str], project_root=project_root, workspace_root=user_config.workspace.root, manifest_path=manifest_path, - state_dir=state_dir, - log_dir=log_dir, - cache_dir=cache_dir, - temp_dir=temp_dir, + state_dir=layout.state_dir, + log_dir=layout.log_dir, + cache_dir=layout.cache_dir, + temp_dir=layout.temp_dir, log_file=log_file, config=config, environment=environment, @@ -222,21 +219,6 @@ def _create_context(self, standard: dict[str, Any], sensitive_options: set[str], ) -def _create_runtime_directory(path: Path, cache_root: Path) -> None: - try: - path.mkdir(parents=True, exist_ok=True) - except OSError as exc: - raise RuntimeError(_runtime_directory_error(path, cache_root, exc)) from exc - - -def _runtime_directory_error(path: Path, cache_root: Path, exc: OSError) -> str: - return ( - f"Unable to create Base runtime directory '{path}': {exc}. " - f"Check permissions on that directory. If the Base cache root '{cache_root}' is unusable, " - "set BASE_CACHE_DIR to a writable directory." - ) - - def run_app(app: App, argv: list[str] | None = None) -> int: try: click = _require_click() @@ -401,33 +383,3 @@ def group_wrapper(context: Any, **kwargs: Any) -> None: context.obj = obj return group_wrapper - - -def _prune_log_files( - log_dir: Path, - current_log_file: Path, - max_log_files: int, - logger: logging.Logger, -) -> None: - candidates: list[tuple[str, Path]] = [] - for path in log_dir.glob("*.log"): - if _same_path(path, current_log_file): - continue - candidates.append((path.name, path)) - - excess_count = len(candidates) + 1 - max_log_files - if excess_count <= 0: - return - - for _, path in sorted(candidates)[:excess_count]: - try: - path.unlink() - except OSError as exc: - logger.warning("Could not prune log file '%s': %s", path, exc) - - -def _same_path(left: Path, right: Path) -> bool: - try: - return left.resolve() == right.resolve() - except OSError: - return left.absolute() == right.absolute() diff --git a/lib/python/base_cli/tests/test_app_runtime_boundary.py b/lib/python/base_cli/tests/test_app_runtime_boundary.py new file mode 100644 index 00000000..16a300e2 --- /dev/null +++ b/lib/python/base_cli/tests/test_app_runtime_boundary.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +import importlib +import importlib.util +from pathlib import Path + +import base_cli +from base_cli import app + + +def test_runtime_layout_names_base_cli_directories() -> None: + assert importlib.util.find_spec("base_cli._runtime") is not None + runtime = importlib.import_module("base_cli._runtime") + layout = runtime.runtime_layout(Path("/tmp/base-cache"), "demo", "run-123") + + assert layout.state_dir == Path("/tmp/base-cache/cli/demo") + assert layout.log_dir == Path("/tmp/base-cache/cli/demo/logs") + assert layout.cache_dir == Path("/tmp/base-cache/cli/demo/cache") + assert layout.temp_dir == Path("/tmp/base-cache/cli/demo/tmp/run-123") + + +def test_runtime_helpers_stay_out_of_public_api() -> None: + assert "_runtime" not in base_cli.__all__ + assert "runtime_layout" not in base_cli.__all__ + + +def test_runtime_directory_helpers_are_split_from_app_module() -> None: + app_source = Path(app.__file__).read_text(encoding="utf-8") + + assert "def runtime_layout" not in app_source + assert "def create_runtime_directory" not in app_source + assert "def prune_log_files" not in app_source