From bb2f711e5ef9ead9f04008bfd2177cad30dc2bf7 Mon Sep 17 00:00:00 2001 From: ulises-jeremias Date: Thu, 16 Jul 2026 03:03:18 -0300 Subject: [PATCH] feat(core): git download cache with refresh modes and offline support Closes #26 Co-authored-by: Cursor --- .../src/create_python_app_core/__init__.py | 10 ++ .../src/create_python_app_core/git_cache.py | 156 ++++++++++++++++++ .../tests/test_git_cache.py | 43 +++++ 3 files changed, 209 insertions(+) create mode 100644 packages/create-python-app-core/src/create_python_app_core/git_cache.py create mode 100644 packages/create-python-app-core/tests/test_git_cache.py diff --git a/packages/create-python-app-core/src/create_python_app_core/__init__.py b/packages/create-python-app-core/src/create_python_app_core/__init__.py index 0d5e45c..44b63d0 100644 --- a/packages/create-python-app-core/src/create_python_app_core/__init__.py +++ b/packages/create-python-app-core/src/create_python_app_core/__init__.py @@ -19,6 +19,12 @@ PackageManagerFallbackError, ScaffoldAbortedError, ) +from create_python_app_core.git_cache import ( + CacheMeta, + download_repository, + read_cache_meta, + write_cache_meta, +) from create_python_app_core.paths import default_cache_dir, resolve_source __all__ = [ @@ -32,6 +38,10 @@ "NON_EMPTY_DIR_ERROR_CODE", "default_cache_dir", "resolve_source", + "download_repository", + "CacheMeta", + "read_cache_meta", + "write_cache_meta", "CPA_USER_AGENT", "check_for_latest_version", "check_python_version", diff --git a/packages/create-python-app-core/src/create_python_app_core/git_cache.py b/packages/create-python-app-core/src/create_python_app_core/git_cache.py new file mode 100644 index 0000000..15b9a1c --- /dev/null +++ b/packages/create-python-app-core/src/create_python_app_core/git_cache.py @@ -0,0 +1,156 @@ +"""Git download + on-disk cache (~/.cache/cpa) with refresh modes.""" + +from __future__ import annotations + +import json +import os +import shutil +import subprocess +import time +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Literal + +from create_python_app_core.errors import CpaError +from create_python_app_core.paths import ResolvedSource, default_cache_dir + +RefreshMode = Literal["always", "stale", "manual"] + + +@dataclass +class CacheMeta: + url: str + ref: str | None + fetched_at: float + commit: str | None = None + + +def resolve_refresh_mode(explicit: RefreshMode | None = None) -> RefreshMode: + if explicit: + return explicit + env = os.environ.get("CPA_REFRESH", "").lower() + if env in ("always", "stale", "manual"): + return env # type: ignore[return-value] + return "stale" + + +def refresh_after_hours() -> float: + raw = os.environ.get("CPA_REFRESH_AFTER_HOURS", "24") + try: + return float(raw) + except ValueError: + return 24.0 + + +def _cache_key(source: ResolvedSource) -> str: + safe = source.url.replace("://", "_").replace("/", "_").replace(":", "_") + ref = source.ref or "default" + return f"{safe}__{ref}" + + +def cache_entry_dir(source: ResolvedSource, cache_root: Path | None = None) -> Path: + root = cache_root or default_cache_dir() + return root / "repos" / _cache_key(source) + + +def meta_path(entry: Path) -> Path: + return entry / ".cpa-cache.json" + + +def write_cache_meta(entry: Path, meta: CacheMeta) -> None: + entry.mkdir(parents=True, exist_ok=True) + meta_path(entry).write_text( + json.dumps(asdict(meta), indent=2) + "\n", encoding="utf-8" + ) + + +def read_cache_meta(entry: Path) -> CacheMeta | None: + path = meta_path(entry) + if not path.is_file(): + return None + data = json.loads(path.read_text(encoding="utf-8")) + return CacheMeta(**data) + + +def _should_refresh(meta: CacheMeta | None, mode: RefreshMode) -> bool: + if mode == "always": + return True + if mode == "manual": + return False + # stale + if meta is None: + return True + age_h = (time.time() - meta.fetched_at) / 3600.0 + return age_h >= refresh_after_hours() + + +def _run_git(args: list[str], *, cwd: Path | None = None) -> str: + if os.environ.get("CPA_SKIP_GIT") == "1": + raise CpaError("Git disabled via CPA_SKIP_GIT=1", code="CPA_SKIP_GIT") + try: + out = subprocess.check_output( + ["git", *args], + cwd=str(cwd) if cwd else None, + stderr=subprocess.STDOUT, + text=True, + ) + return out.strip() + except subprocess.CalledProcessError as exc: + raise CpaError( + f"git {' '.join(args)} failed: {exc.output}", code="CPA_GIT" + ) from exc + except FileNotFoundError as exc: + raise CpaError("git executable not found", code="CPA_GIT") from exc + + +def download_repository( + source: ResolvedSource, + *, + offline: bool = False, + refresh: RefreshMode | None = None, + cache_root: Path | None = None, +) -> Path: + """Clone or refresh a repo into the cache; return the entry directory.""" + if source.kind == "file": + if source.local_path is None or not source.local_path.exists(): + raise CpaError(f"file source not found: {source.url}", code="CPA_FILE") + return source.local_path + + entry = cache_entry_dir(source, cache_root) + mode = resolve_refresh_mode(refresh) + meta = read_cache_meta(entry) + + if offline: + if not entry.exists(): + raise CpaError( + f"offline mode: cache miss for {source.url}", + code="CPA_OFFLINE", + ) + return entry + + if entry.exists() and not _should_refresh(meta, mode): + return entry + + if entry.exists() and (entry / ".git").is_dir() and mode != "manual": + _run_git(["fetch", "--all", "--tags"], cwd=entry) + if source.ref: + _run_git(["checkout", source.ref], cwd=entry) + commit = _run_git(["rev-parse", "HEAD"], cwd=entry) + else: + if entry.exists(): + shutil.rmtree(entry) + entry.mkdir(parents=True, exist_ok=True) + clone_args = ["clone", "--depth", "1"] + if source.ref: + clone_args.extend(["--branch", source.ref]) + clone_args.extend([source.url, str(entry)]) + _run_git(clone_args) + commit = _run_git(["rev-parse", "HEAD"], cwd=entry) + + write_cache_meta( + entry, + CacheMeta( + url=source.url, ref=source.ref, fetched_at=time.time(), commit=commit + ), + ) + return entry diff --git a/packages/create-python-app-core/tests/test_git_cache.py b/packages/create-python-app-core/tests/test_git_cache.py new file mode 100644 index 0000000..dd0afa0 --- /dev/null +++ b/packages/create-python-app-core/tests/test_git_cache.py @@ -0,0 +1,43 @@ +import json +import time +from pathlib import Path + +import pytest +from create_python_app_core.errors import CpaError +from create_python_app_core.git_cache import ( + CacheMeta, + download_repository, + read_cache_meta, + write_cache_meta, +) +from create_python_app_core.paths import ResolvedSource + + +def test_file_source_returns_path(tmp_path: Path) -> None: + src = ResolvedSource(kind="file", url=f"file://{tmp_path}", local_path=tmp_path) + assert download_repository(src) == tmp_path + + +def test_offline_miss_raises(tmp_path: Path) -> None: + src = ResolvedSource(kind="github", url="https://github.com/org/repo") + with pytest.raises(CpaError) as ei: + download_repository(src, offline=True, cache_root=tmp_path) + assert ei.value.code == "CPA_OFFLINE" + + +def test_meta_roundtrip(tmp_path: Path) -> None: + entry = tmp_path / "e" + meta = CacheMeta(url="u", ref="main", fetched_at=time.time(), commit="abc") + write_cache_meta(entry, meta) + loaded = read_cache_meta(entry) + assert loaded is not None + assert loaded.url == "u" + assert json.loads((entry / ".cpa-cache.json").read_text())["ref"] == "main" + + +def test_skip_git_env(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + monkeypatch.setenv("CPA_SKIP_GIT", "1") + src = ResolvedSource(kind="github", url="https://github.com/org/repo") + with pytest.raises(CpaError) as ei: + download_repository(src, cache_root=tmp_path, refresh="always") + assert ei.value.code == "CPA_SKIP_GIT"