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