Coverage for src/secchi/cache.py: 82%

1"""Daily package-response cache for registry/GitHub API data.""" 

2 

3from __future__ import annotations 

4 

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 

12 

13from secchi.models import ( 

14 Dependency, 

15 DownloadCounts, 

16 DownloadTrendPoint, 

17 GitHubIssueEvent, 

18 GitHubStats, 

19 MetricTimelinePoint, 

20 PackageInfo, 

21 Registry, 

22 ReleaseFile, 

23 ReverseDependency, 

24 Version, 

25) 

26from secchi.schema import CACHE_SCHEMA_VERSION 

27from secchi.schemas import CacheEnvelope 

28 

29 

30def cache_root() -> Path: 

31 """XDG_CACHE_HOME/secchi, else ~/.cache/secchi.""" 

32 if base := os.environ.get("XDG_CACHE_HOME", ""): 

33 return Path(base) / "secchi" 

34 return Path.home() / ".cache" / "secchi" 

35 

36 

37def package_cache_path(key: str, *, root: Path | None = None) -> Path: 

38 safe = key.replace("/", "_").replace(":", "__") 

39 return (root or cache_root()) / "packages" / f"{safe}.json" 

40 

41 

42def load_package_cache( 

43 key: str, 

44 *, 

45 root: Path | None = None, 

46 now: Callable[[], datetime] | None = None, 

47) -> tuple[PackageInfo, datetime] | None: 

48 path = package_cache_path(key, root=root) 

49 if not path.exists(): 

50 return None 

51 try: 

52 raw = json.loads(path.read_text()) 

53 if not isinstance(raw, dict): 

54 return None 

55 schema_version = raw.get("schema_version", 0) 

56 if not isinstance(schema_version, int) or schema_version > CACHE_SCHEMA_VERSION: 

57 return None 

58 envelope = CacheEnvelope.model_validate( 

59 {**raw, "schema_version": schema_version} 

60 ) 

61 fetched_at = envelope.fetched_at 

62 today = (now or (lambda: datetime.now().astimezone()))().date() 

63 if fetched_at.astimezone().date() != today: 

64 return None 

65 return _decode_package_info(envelope.package), fetched_at 

66 except (KeyError, TypeError, ValueError, json.JSONDecodeError, OSError): 

67 return None 

68 

69 

70def save_package_cache( 

71 key: str, 

72 info: PackageInfo, 

73 fetched_at: datetime, 

74 *, 

75 root: Path | None = None, 

76) -> None: 

77 path = package_cache_path(key, root=root) 

78 payload = CacheEnvelope( 

79 schema_version=CACHE_SCHEMA_VERSION, 

80 fetched_at=fetched_at, 

81 package=_encode(asdict(info)), 

82 ).model_dump(mode="json") 

83 try: 

84 path.parent.mkdir(parents=True, exist_ok=True) 

85 path.write_text(json.dumps(payload, indent=2, sort_keys=True)) 

86 except OSError: 

87 pass 

88 

89 

90def _encode(value: Any) -> Any: 

91 if isinstance(value, datetime): 

92 return value.isoformat() 

93 if isinstance(value, Registry): 

94 return value.value 

95 if isinstance(value, dict): 

96 return {str(k): _encode(v) for k, v in value.items()} 

97 if isinstance(value, list): 

98 return [_encode(v) for v in value] 

99 return value 

100 

101 

102def _decode_package_info(raw: dict[str, Any]) -> PackageInfo: 

103 info = PackageInfo( 

104 name=raw.get("name", ""), 

105 registry=Registry(raw.get("registry", "pypi")), 

106 source_registries=[Registry(r) for r in raw.get("source_registries", []) if r], 

107 description=raw.get("description", ""), 

108 author=raw.get("author", ""), 

109 license=raw.get("license", ""), 

110 homepage=raw.get("homepage", ""), 

111 repository_url=raw.get("repository_url", ""), 

112 documentation_url=raw.get("documentation_url", ""), 

113 latest_version=raw.get("latest_version", ""), 

114 latest_release_date=_parse_datetime(raw.get("latest_release_date")), 

115 total_downloads=raw.get("total_downloads", 0), 

116 package_kind=raw.get("package_kind", ""), 

117 ) 

118 info.download_counts = _decode_download_counts(raw.get("download_counts", {})) 

119 info.github_stats = _decode_github_stats(raw.get("github_stats", {})) 

120 info.versions = [_decode_version(v) for v in raw.get("versions", [])] 

121 info.dependencies = [_decode_dependency(d) for d in raw.get("dependencies", [])] 

122 info.download_trend = [ 

123 DownloadTrendPoint(date=p.get("date", ""), count=p.get("count", 0)) 

124 for p in raw.get("download_trend", []) 

125 ] 

126 info.release_notes = raw.get("release_notes", "") 

127 info.latest_release_files = [ 

128 ReleaseFile( 

129 packagetype=f.get("packagetype", ""), 

130 size=f.get("size", 0), 

131 filename=f.get("filename", ""), 

132 ) 

133 for f in raw.get("latest_release_files", []) 

134 ] 

135 info.version_downloads_recent = { 

136 _decode_key(k): v for k, v in raw.get("version_downloads_recent", {}).items() 

137 } 

138 info.reverse_dependencies = [ 

139 ReverseDependency(name=d.get("name", ""), downloads=d.get("downloads", 0)) 

140 for d in raw.get("reverse_dependencies", []) 

141 ] 

142 info.reverse_dependency_count = raw.get("reverse_dependency_count") 

143 info.reverse_dependency_monthly_growth = raw.get( 

144 "reverse_dependency_monthly_growth" 

145 ) 

146 info.health_history = [ 

147 MetricTimelinePoint(label=p.get("label", ""), value=p.get("value", 0)) 

148 for p in raw.get("health_history", []) 

149 ] 

150 info.github_issue_events = [ 

151 _decode_issue_event(e) for e in raw.get("github_issue_events", []) 

152 ] 

153 return info 

154 

155 

156def _decode_download_counts(raw: dict[str, Any]) -> DownloadCounts: 

157 return DownloadCounts( 

158 today=raw.get("today", 0), 

159 week=raw.get("week", 0), 

160 month=raw.get("month", 0), 

161 ) 

162 

163 

164def _decode_github_stats(raw: dict[str, Any]) -> GitHubStats: 

165 return GitHubStats( 

166 stars=raw.get("stars", 0), 

167 forks=raw.get("forks", 0), 

168 open_issues=raw.get("open_issues", 0), 

169 created_at=_parse_datetime(raw.get("created_at")), 

170 pushed_at=_parse_datetime(raw.get("pushed_at")), 

171 has_ci=raw.get("has_ci", False), 

172 has_readme=raw.get("has_readme", False), 

173 resolved=raw.get("resolved", False), 

174 stars_delta_7d=raw.get("stars_delta_7d"), 

175 open_issues_delta_7d=raw.get("open_issues_delta_7d"), 

176 ) 

177 

178 

179def _decode_version(raw: dict[str, Any]) -> Version: 

180 return Version( 

181 version=raw.get("version", ""), 

182 release_date=_parse_datetime(raw.get("release_date")), 

183 downloads=raw.get("downloads", 0), 

184 is_yanked=raw.get("is_yanked", False), 

185 external_id=raw.get("external_id"), 

186 size_bytes=raw.get("size_bytes"), 

187 ) 

188 

189 

190def _decode_dependency(raw: dict[str, Any]) -> Dependency: 

191 return Dependency( 

192 name=raw.get("name", ""), 

193 requirement=raw.get("requirement", ""), 

194 optional=raw.get("optional", False), 

195 ) 

196 

197 

198def _decode_issue_event(raw: dict[str, Any]) -> GitHubIssueEvent: 

199 return GitHubIssueEvent( 

200 number=raw.get("number", 0), 

201 title=raw.get("title", ""), 

202 is_pull_request=raw.get("is_pull_request", False), 

203 created_at=_parse_datetime(raw.get("created_at")) or datetime.min, 

204 closed_at=_parse_datetime(raw.get("closed_at")), 

205 url=raw.get("url", ""), 

206 ) 

207 

208 

209def _parse_datetime(raw: Any) -> datetime | None: 

210 if not raw: 

211 return None 

212 try: 

213 return datetime.fromisoformat(raw) 

214 except (TypeError, ValueError): 

215 return None 

216 

217 

218def _decode_key(raw: str) -> int | str: 

219 try: 

220 return int(raw) 

221 except ValueError: 

222 return raw