Coverage for src/secchi/api/crates.py: 59%
169 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-14 23:28 +0000
1"""crates.io registry adapter — direct REST API."""
3from __future__ import annotations
5import asyncio
6from datetime import UTC, datetime
8import httpx
10from secchi.api.base import AdapterBase, RegistryAdapter
11from secchi.models import (
12 Dependency,
13 DownloadCounts,
14 DownloadTrendPoint,
15 PackageInfo,
16 Registry,
17 ReleaseFile,
18 ReverseDependency,
19 SearchResult,
20 Version,
21)
23CRATES_API = "https://crates.io/api/v1"
25# Aware sort sentinel for versions with no parsed release date — release_date
26# is always tz-aware when present, and a naive datetime.min would raise
27# TypeError when compared against it, dropping the whole version list.
28_MIN_DATETIME = datetime.min.replace(tzinfo=UTC)
30_HEADERS = {
31 "User-Agent": "secchi (https://github.com/kannandreams/secchi)",
32 "Accept": "application/json",
33}
35# crates.io category slugs / names that indicate a CLI tool.
36_CLI_CATEGORIES = {
37 "command-line-utilities",
38 "command-line-interface",
39 "development-tools::cargo-plugins",
40}
43class CratesAdapter(AdapterBase, RegistryAdapter):
44 default_headers = _HEADERS
46 @property
47 def registry(self) -> Registry:
48 return Registry.CRATES
50 async def fetch_package(self, name: str) -> PackageInfo:
51 async with self._client_scope() as client:
52 resp = await client.get(f"{CRATES_API}/crates/{name}")
53 resp.raise_for_status()
54 payload = resp.json()
55 data = payload["crate"]
57 latest_version = data.get("max_stable_version", "")
58 if not latest_version:
59 latest_version = data.get("max_version", "")
61 total_downloads = data.get("downloads", 0)
63 # package_kind from categories (best-effort; never blocks the fetch)
64 kind = "Library"
65 for cat in payload.get("categories", []):
66 slug = (cat.get("slug") or "").lower()
67 if slug in _CLI_CATEGORIES:
68 kind = "CLI"
69 break
71 # synthetic release file for the latest version's crate size;
72 # license lives on the version object, not the top-level crate.
73 release_files: list[ReleaseFile] = []
74 license = data.get("license", "") or ""
75 for ver in payload.get("versions", []):
76 if ver.get("num") == latest_version:
77 release_files.append(
78 ReleaseFile(
79 packagetype="crate",
80 size=ver.get("crate_size") or 0,
81 filename=f"{name}-{latest_version}.crate",
82 )
83 )
84 license = license or ver.get("license", "") or ""
85 break
86 if not license:
87 for ver in payload.get("versions", []):
88 if ver.get("license"):
89 license = ver["license"]
90 break
92 return PackageInfo(
93 name=data["name"],
94 registry=Registry.CRATES,
95 description=data.get("description", ""),
96 license=license,
97 homepage=data.get("homepage", ""),
98 repository_url=data.get("repository", ""),
99 documentation_url=data.get("documentation", ""),
100 latest_version=latest_version,
101 latest_release_date=_parse_date(data.get("updated_at")),
102 total_downloads=total_downloads,
103 package_kind=kind,
104 latest_release_files=release_files,
105 )
107 async def fetch_versions(self, name: str) -> list[Version]:
108 async with self._client_scope() as client:
109 resp = await client.get(f"{CRATES_API}/crates/{name}")
110 resp.raise_for_status()
111 data = resp.json()
113 versions: list[Version] = []
114 for ver_data in data.get("versions", []):
115 versions.append(
116 Version(
117 version=ver_data["num"],
118 release_date=_parse_date(ver_data.get("created_at")),
119 downloads=ver_data.get("downloads", 0),
120 is_yanked=ver_data.get("yanked", False),
121 external_id=ver_data.get("id"),
122 size_bytes=ver_data.get("crate_size"),
123 )
124 )
126 versions.sort(key=lambda v: v.release_date or _MIN_DATETIME, reverse=True)
127 return versions
129 async def fetch_dependencies(self, name: str, version: str) -> list[Dependency]:
130 async with self._client_scope() as client:
131 resp = await client.get(
132 f"{CRATES_API}/crates/{name}/{version}/dependencies"
133 )
134 resp.raise_for_status()
135 data = resp.json()
137 deps: list[Dependency] = []
138 for dep in data.get("dependencies", []):
139 deps.append(
140 Dependency(
141 name=dep["crate_id"],
142 requirement=dep.get("req", "*"),
143 optional=dep.get("optional", False),
144 )
145 )
146 return deps
148 async def fetch_download_trend(
149 self, name: str, days: int = 30
150 ) -> list[DownloadTrendPoint]:
151 async with self._client_scope() as client:
152 resp = await client.get(f"{CRATES_API}/crates/{name}/downloads")
153 resp.raise_for_status()
154 data = resp.json()
156 daily: dict[str, int] = {}
157 for entry in data.get("version_downloads", []):
158 date = entry["date"]
159 daily[date] = daily.get(date, 0) + entry.get("downloads", 0)
161 for entry in data.get("meta", {}).get("extra_downloads", []):
162 date = entry["date"]
163 daily[date] = daily.get(date, 0) + entry.get("downloads", 0)
165 points = [
166 DownloadTrendPoint(date=date, count=count)
167 for date, count in sorted(daily.items())
168 ]
169 return points[-days:] if len(points) > days else points
171 async def fetch_download_counts(self, name: str) -> DownloadCounts:
172 trend = await self.fetch_download_trend(name, days=30)
173 if not trend:
174 return DownloadCounts()
176 today = trend[-1].count if trend else 0
177 week = sum(p.count for p in trend[-7:])
178 month = sum(p.count for p in trend)
179 return DownloadCounts(today=today, week=week, month=month)
181 async def fetch_version_download_breakdown(self, name: str) -> dict[int | str, int]:
182 """Sum version_downloads per crates.io numeric version id.
184 Real per-version signal — the numeric id joins to Version.external_id.
185 """
186 async with self._client_scope() as client:
187 try:
188 resp = await client.get(f"{CRATES_API}/crates/{name}/downloads")
189 resp.raise_for_status()
190 data = resp.json()
191 except httpx.HTTPError:
192 return {}
193 totals: dict[int | str, int] = {}
194 for entry in data.get("version_downloads", []):
195 vid = entry.get("version")
196 if vid is None:
197 continue
198 totals[vid] = totals.get(vid, 0) + entry.get("downloads", 0)
199 return totals
201 async def fetch_reverse_dependencies(
202 self, name: str, limit: int = 5
203 ) -> list[ReverseDependency]:
204 """Reverse dependencies ranked by each dependent's real total downloads."""
205 async with self._client_scope() as client:
206 try:
207 resp = await client.get(
208 f"{CRATES_API}/crates/{name}/reverse_dependencies",
209 params={"per_page": "30", "sort": "downloads"},
210 )
211 resp.raise_for_status()
212 data = resp.json()
213 except httpx.HTTPError:
214 return []
216 # Join dependency edges -> version -> owning crate name.
217 versions = {v.get("id"): v.get("crate") for v in data.get("versions", [])}
218 names: list[str] = []
219 seen: set[str] = set()
220 for dep in data.get("dependencies", []):
221 crate_name = versions.get(dep.get("version_id"))
222 if crate_name and crate_name not in seen:
223 seen.add(crate_name)
224 names.append(crate_name)
225 if len(names) >= 10:
226 break
228 async def _downloads(crate_name: str) -> ReverseDependency:
229 try:
230 r = await client.get(f"{CRATES_API}/crates/{crate_name}")
231 r.raise_for_status()
232 dl = r.json()["crate"].get("downloads", 0)
233 except (httpx.HTTPError, KeyError):
234 dl = 0
235 return ReverseDependency(name=crate_name, downloads=dl)
237 results = await asyncio.gather(*[_downloads(n) for n in names])
238 results.sort(key=lambda r: r.downloads, reverse=True)
239 return results[:limit]
241 async def fetch_reverse_dependency_count(self, name: str) -> int | None:
242 async with self._client_scope() as client:
243 try:
244 resp = await client.get(
245 f"{CRATES_API}/crates/{name}/reverse_dependencies",
246 params={"per_page": "1"},
247 )
248 resp.raise_for_status()
249 data = resp.json()
250 except httpx.HTTPError:
251 return None
253 return _reverse_dependency_total(data)
255 async def fetch_release_notes(self, name: str, version: str) -> str:
256 return "" # fetched via GitHub in utils
258 async def search(self, query: str, limit: int = 10) -> list[SearchResult]:
259 async with self._client_scope() as client:
260 try:
261 response = await client.get(
262 f"{CRATES_API}/crates",
263 params={"q": query, "per_page": limit},
264 )
265 response.raise_for_status()
266 except httpx.HTTPError:
267 return []
268 results: list[SearchResult] = []
269 for crate in response.json().get("crates", [])[:limit]:
270 name = crate.get("id", crate.get("name", ""))
271 results.append(
272 SearchResult(
273 name=name,
274 registry=Registry.CRATES,
275 version=crate.get("max_version", ""),
276 description=crate.get("description", "") or "",
277 url=f"https://crates.io/crates/{name}",
278 score=float(crate.get("recent_downloads", 0) or 0),
279 exact=name.lower() == query.lower(),
280 )
281 )
282 return results
285def _parse_date(raw: str | None) -> datetime | None:
286 if not raw:
287 return None
288 try:
289 raw_clean = raw.replace("Z", "+00:00")
290 return datetime.fromisoformat(raw_clean)
291 except (ValueError, TypeError):
292 return None
295def _reverse_dependency_total(data: dict) -> int | None:
296 meta = data.get("meta", {})
297 for key in ("total", "total_count", "count"):
298 value = meta.get(key)
299 if isinstance(value, int):
300 return value
301 if isinstance(value, str) and value.isdigit():
302 return int(value)
303 return None