Coverage for src/secchi/cache.py: 81%
134 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-14 23:28 +0000
1"""Daily package-response cache for registry/GitHub API data."""
3from __future__ import annotations
5import json
6import os
7from collections.abc import Callable
8from dataclasses import asdict
9from datetime import datetime
10from pathlib import Path
11from typing import Any
12from urllib.parse import quote
14from secchi.models import (
15 AdvisoryReference,
16 Dependency,
17 DownloadCounts,
18 DownloadTrendPoint,
19 GitHubIssueEvent,
20 GitHubStats,
21 MetricTimelinePoint,
22 PackageInfo,
23 Registry,
24 ReleaseFile,
25 ReverseDependency,
26 SecurityAdvisory,
27 Version,
28)
29from secchi.schema import CACHE_SCHEMA_VERSION, SECURITY_CACHE_SCHEMA_VERSION
30from secchi.schemas import CacheEnvelope
33def cache_root() -> Path:
34 """XDG_CACHE_HOME/secchi, else ~/.cache/secchi."""
35 if base := os.environ.get("XDG_CACHE_HOME", ""):
36 return Path(base) / "secchi"
37 return Path.home() / ".cache" / "secchi"
40def _safe_cache_key(key: str) -> str:
41 """Percent-encode a cache key into a single, collision-free filename.
43 ``quote(key, safe="")`` escapes every byte outside the URL-unreserved
44 set, so ``/`` and ``\\`` (path separators on POSIX/Windows) and ``:``
45 (invalid on Windows) can never reach the filesystem — the result is
46 always exactly one path component, ruling out traversal outside the
47 cache root. Because the encoding is injective, distinct keys can no
48 longer collide onto the same file (unlike a lossy character replace,
49 where e.g. ``a/b_c`` and ``a_b/c`` previously mapped to one file).
50 """
51 return quote(key, safe="")
54def package_cache_path(key: str, *, root: Path | None = None) -> Path:
55 return (root or cache_root()) / "packages" / f"{_safe_cache_key(key)}.json"
58def security_cache_path(key: str, *, root: Path | None = None) -> Path:
59 return (root or cache_root()) / "security" / f"{_safe_cache_key(key)}.json"
62def load_security_cache(
63 key: str,
64 package_version: str,
65 *,
66 root: Path | None = None,
67 now: Callable[[], datetime] | None = None,
68) -> tuple[list[SecurityAdvisory], datetime] | None:
69 path = security_cache_path(key, root=root)
70 if not path.exists():
71 return None
72 try:
73 raw = json.loads(path.read_text())
74 if not isinstance(raw, dict):
75 return None
76 if raw.get("schema_version") != SECURITY_CACHE_SCHEMA_VERSION:
77 return None
78 if raw.get("package_version", "") != package_version:
79 return None
80 fetched_at = _parse_datetime(raw.get("fetched_at"))
81 if fetched_at is None:
82 return None
83 today = (now or (lambda: datetime.now().astimezone()))().date()
84 if fetched_at.astimezone().date() != today:
85 return None
86 advisories = raw.get("advisories", [])
87 if not isinstance(advisories, list):
88 return None
89 if not all(isinstance(item, dict) for item in advisories):
90 return None
91 return [_decode_advisory(item) for item in advisories], fetched_at
92 except (KeyError, TypeError, ValueError, json.JSONDecodeError, OSError):
93 return None
96def save_security_cache(
97 key: str,
98 package_version: str,
99 advisories: list[SecurityAdvisory],
100 fetched_at: datetime,
101 *,
102 root: Path | None = None,
103) -> None:
104 path = security_cache_path(key, root=root)
105 payload = {
106 "schema_version": SECURITY_CACHE_SCHEMA_VERSION,
107 "fetched_at": fetched_at.isoformat(),
108 "package_version": package_version,
109 "advisories": _encode([asdict(advisory) for advisory in advisories]),
110 }
111 try:
112 path.parent.mkdir(parents=True, exist_ok=True)
113 path.write_text(json.dumps(payload, indent=2, sort_keys=True))
114 except OSError:
115 pass
118def load_package_cache(
119 key: str,
120 *,
121 root: Path | None = None,
122 now: Callable[[], datetime] | None = None,
123) -> tuple[PackageInfo, datetime] | None:
124 path = package_cache_path(key, root=root)
125 if not path.exists():
126 return None
127 try:
128 raw = json.loads(path.read_text())
129 if not isinstance(raw, dict):
130 return None
131 schema_version = raw.get("schema_version", 0)
132 if not isinstance(schema_version, int) or schema_version > CACHE_SCHEMA_VERSION:
133 return None
134 envelope = CacheEnvelope.model_validate(
135 {**raw, "schema_version": schema_version}
136 )
137 fetched_at = envelope.fetched_at
138 today = (now or (lambda: datetime.now().astimezone()))().date()
139 if fetched_at.astimezone().date() != today:
140 return None
141 return _decode_package_info(envelope.package), fetched_at
142 except (KeyError, TypeError, ValueError, json.JSONDecodeError, OSError):
143 return None
146def save_package_cache(
147 key: str,
148 info: PackageInfo,
149 fetched_at: datetime,
150 *,
151 root: Path | None = None,
152) -> None:
153 path = package_cache_path(key, root=root)
154 payload = CacheEnvelope(
155 schema_version=CACHE_SCHEMA_VERSION,
156 fetched_at=fetched_at,
157 package=_encode(asdict(info)),
158 ).model_dump(mode="json")
159 try:
160 path.parent.mkdir(parents=True, exist_ok=True)
161 path.write_text(json.dumps(payload, indent=2, sort_keys=True))
162 except OSError:
163 pass
166def _encode(value: Any) -> Any:
167 if isinstance(value, datetime):
168 return value.isoformat()
169 if isinstance(value, Registry):
170 return value.value
171 if isinstance(value, dict):
172 return {str(k): _encode(v) for k, v in value.items()}
173 if isinstance(value, list):
174 return [_encode(v) for v in value]
175 return value
178def _decode_package_info(raw: dict[str, Any]) -> PackageInfo:
179 info = PackageInfo(
180 name=raw.get("name", ""),
181 registry=Registry(raw.get("registry", "pypi")),
182 source_registries=[Registry(r) for r in raw.get("source_registries", []) if r],
183 description=raw.get("description", ""),
184 author=raw.get("author", ""),
185 license=raw.get("license", ""),
186 homepage=raw.get("homepage", ""),
187 repository_url=raw.get("repository_url", ""),
188 documentation_url=raw.get("documentation_url", ""),
189 latest_version=raw.get("latest_version", ""),
190 latest_release_date=_parse_datetime(raw.get("latest_release_date")),
191 total_downloads=raw.get("total_downloads", 0),
192 package_kind=raw.get("package_kind", ""),
193 )
194 info.download_counts = _decode_download_counts(raw.get("download_counts", {}))
195 info.github_stats = _decode_github_stats(raw.get("github_stats", {}))
196 info.versions = [_decode_version(v) for v in raw.get("versions", [])]
197 info.dependencies = [_decode_dependency(d) for d in raw.get("dependencies", [])]
198 info.download_trend = [
199 DownloadTrendPoint(date=p.get("date", ""), count=p.get("count", 0))
200 for p in raw.get("download_trend", [])
201 ]
202 info.release_notes = raw.get("release_notes", "")
203 info.latest_release_files = [
204 ReleaseFile(
205 packagetype=f.get("packagetype", ""),
206 size=f.get("size", 0),
207 filename=f.get("filename", ""),
208 )
209 for f in raw.get("latest_release_files", [])
210 ]
211 info.version_downloads_recent = {
212 _decode_key(k): v for k, v in raw.get("version_downloads_recent", {}).items()
213 }
214 info.reverse_dependencies = [
215 ReverseDependency(name=d.get("name", ""), downloads=d.get("downloads", 0))
216 for d in raw.get("reverse_dependencies", [])
217 ]
218 info.reverse_dependency_count = raw.get("reverse_dependency_count")
219 info.reverse_dependency_monthly_growth = raw.get(
220 "reverse_dependency_monthly_growth"
221 )
222 info.health_history = [
223 MetricTimelinePoint(label=p.get("label", ""), value=p.get("value", 0))
224 for p in raw.get("health_history", [])
225 ]
226 info.github_issue_events = [
227 _decode_issue_event(e) for e in raw.get("github_issue_events", [])
228 ]
229 info.security_advisories = [
230 _decode_advisory(a) for a in raw.get("security_advisories", [])
231 ]
232 return info
235def _decode_download_counts(raw: dict[str, Any]) -> DownloadCounts:
236 return DownloadCounts(
237 today=raw.get("today", 0),
238 week=raw.get("week", 0),
239 month=raw.get("month", 0),
240 )
243def _decode_github_stats(raw: dict[str, Any]) -> GitHubStats:
244 return GitHubStats(
245 stars=raw.get("stars", 0),
246 forks=raw.get("forks", 0),
247 open_issues=raw.get("open_issues", 0),
248 created_at=_parse_datetime(raw.get("created_at")),
249 pushed_at=_parse_datetime(raw.get("pushed_at")),
250 has_ci=raw.get("has_ci", False),
251 has_readme=raw.get("has_readme", False),
252 resolved=raw.get("resolved", False),
253 stars_delta_7d=raw.get("stars_delta_7d"),
254 open_issues_delta_7d=raw.get("open_issues_delta_7d"),
255 )
258def _decode_version(raw: dict[str, Any]) -> Version:
259 return Version(
260 version=raw.get("version", ""),
261 release_date=_parse_datetime(raw.get("release_date")),
262 downloads=raw.get("downloads", 0),
263 is_yanked=raw.get("is_yanked", False),
264 external_id=raw.get("external_id"),
265 size_bytes=raw.get("size_bytes"),
266 )
269def _decode_dependency(raw: dict[str, Any]) -> Dependency:
270 return Dependency(
271 name=raw.get("name", ""),
272 requirement=raw.get("requirement", ""),
273 optional=raw.get("optional", False),
274 )
277def _decode_issue_event(raw: dict[str, Any]) -> GitHubIssueEvent:
278 return GitHubIssueEvent(
279 number=raw.get("number", 0),
280 title=raw.get("title", ""),
281 is_pull_request=raw.get("is_pull_request", False),
282 created_at=_parse_datetime(raw.get("created_at")) or datetime.min,
283 closed_at=_parse_datetime(raw.get("closed_at")),
284 url=raw.get("url", ""),
285 )
288def _decode_advisory(raw: dict[str, Any]) -> SecurityAdvisory:
289 return SecurityAdvisory(
290 id=raw.get("id", ""),
291 summary=raw.get("summary", ""),
292 details=raw.get("details", ""),
293 aliases=raw.get("aliases", []),
294 severity=raw.get("severity", ""),
295 published=_parse_datetime(raw.get("published")),
296 modified=_parse_datetime(raw.get("modified")),
297 fixed_versions=raw.get("fixed_versions", []),
298 references=[
299 AdvisoryReference(type=item.get("type", ""), url=item.get("url", ""))
300 for item in raw.get("references", [])
301 if isinstance(item, dict)
302 ],
303 url=raw.get("url", ""),
304 )
307def _parse_datetime(raw: Any) -> datetime | None:
308 if not raw:
309 return None
310 try:
311 return datetime.fromisoformat(raw)
312 except (TypeError, ValueError):
313 return None
316def _decode_key(raw: str) -> int | str:
317 try:
318 return int(raw)
319 except ValueError:
320 return raw