feat(core): git cache (#26)

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",

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

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"