1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
|
"""Persistent Git catalog cache and auto-refresh TTL (ADR-0003, ADR-0012)."""
from __future__ import annotations
import hashlib
import json
import os
import shutil
import stat
from collections.abc import Mapping
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from pathlib import Path
from lpip.config import default_user_cache_dir
from lpip.gitutils import git_checkout, git_clone, git_fetch
from lpip.models import RefreshMode
AUTO_REFRESH_TTL = timedelta(hours=24)
_METADATA_NAME = "metadata.json"
_LAST_SUCCESS_KEY = "last_success"
@dataclass(frozen=True)
class CatalogCachePaths:
root: Path
repo: Path
metadata: Path
def catalog_identity_hash(url: str, revision: str | None) -> str:
payload = f"{url}\n{revision or ''}".encode()
return hashlib.sha256(payload).hexdigest()
def catalog_cache_paths(
url: str,
revision: str | None,
*,
cache_home: Path | None = None,
) -> CatalogCachePaths:
home = cache_home if cache_home is not None else default_user_cache_dir()
root = home / "catalogs" / catalog_identity_hash(url, revision)
return CatalogCachePaths(root=root, repo=root / "repo", metadata=root / _METADATA_NAME)
def read_last_success(paths: CatalogCachePaths) -> datetime | None:
if not paths.metadata.is_file():
return None
try:
data = json.loads(paths.metadata.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return None
raw = data.get(_LAST_SUCCESS_KEY)
if not isinstance(raw, str) or not raw:
return None
try:
parsed = datetime.fromisoformat(raw)
except ValueError:
return None
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=UTC)
return parsed
def write_last_success(paths: CatalogCachePaths, when: datetime | None = None) -> None:
"""Record last successful clone/fetch. Call only after Git succeeds."""
stamp = when or datetime.now(UTC)
if stamp.tzinfo is None:
stamp = stamp.replace(tzinfo=UTC)
payload: dict[str, object] = {}
if paths.metadata.is_file():
try:
loaded = json.loads(paths.metadata.read_text(encoding="utf-8"))
if isinstance(loaded, dict):
payload = loaded
except (OSError, json.JSONDecodeError):
payload = {}
payload[_LAST_SUCCESS_KEY] = stamp.isoformat()
paths.root.mkdir(parents=True, exist_ok=True)
paths.metadata.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
def needs_auto_refresh(
paths: CatalogCachePaths,
*,
now: datetime | None = None,
) -> bool:
"""True if timestamp is missing or older than 24 hours (ADR-0012)."""
last = read_last_success(paths)
if last is None:
return True
clock = now or datetime.now(UTC)
if clock.tzinfo is None:
clock = clock.replace(tzinfo=UTC)
return clock - last >= AUTO_REFRESH_TTL
def should_fetch(
paths: CatalogCachePaths,
refresh: RefreshMode,
*,
force: bool = False,
now: datetime | None = None,
) -> bool:
"""Whether to fetch an existing checkout. Absent cache is always cloned."""
if force or refresh is RefreshMode.ALWAYS:
return True
if refresh is RefreshMode.NEVER:
return False
return needs_auto_refresh(paths, now=now)
def _repo_ready(repo: Path) -> bool:
git_dir = repo / ".git"
return git_dir.is_dir() or git_dir.is_file()
def ensure_catalog_checkout(
url: str,
revision: str | None,
*,
cache_home: Path | None = None,
refresh: RefreshMode = RefreshMode.AUTO,
force_refresh: bool = False,
now: datetime | None = None,
environ: Mapping[str, str] | None = None,
) -> CatalogCachePaths:
"""Clone if missing; fetch per refresh policy; stamp only on Git success."""
paths = catalog_cache_paths(url, revision, cache_home=cache_home)
if not _repo_ready(paths.repo):
git_clone(url, paths.repo, environ=environ)
write_last_success(paths, now)
if revision:
git_checkout(paths.repo, revision, environ=environ)
return paths
if should_fetch(paths, refresh, force=force_refresh, now=now):
git_fetch(paths.repo, environ=environ)
write_last_success(paths, now)
if revision:
git_checkout(paths.repo, revision, environ=environ)
return paths
def _rmtree(path: Path) -> None:
def _onerror(func: object, name: str, _exc: object) -> None:
os.chmod(name, stat.S_IWRITE)
func(name) # type: ignore[operator]
shutil.rmtree(path, onerror=_onerror)
def purge_catalog_cache(
url: str,
revision: str | None,
*,
cache_home: Path | None = None,
) -> CatalogCachePaths:
"""Remove the cache directory for this catalog identity."""
paths = catalog_cache_paths(url, revision, cache_home=cache_home)
if paths.root.exists():
_rmtree(paths.root)
return paths
|