Coverage for src/secchi/api/pypi.py: 76%
141 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-14 23:28 +0000
1"""PyPI registry adapter using PyPI JSON API."""
3from __future__ import annotations
5import re
6from contextlib import suppress
7from datetime import datetime
8from html import unescape
9from typing import Any
11import httpx
13from secchi.api.base import AdapterBase, RegistryAdapter
14from secchi.diagnostics import DiagnosticStatus, diagnostic_for_http_error
15from secchi.models import (
16 Dependency,
17 DownloadCounts,
18 DownloadTrendPoint,
19 PackageInfo,
20 Registry,
21 ReleaseFile,
22 SearchResult,
23 Version,
24)
26PYPI_JSON = "https://pypi.org/pypi"
27PYPI_STATS = "https://pypistats.org/api"
30def _derive_kind(classifiers: list[str]) -> str:
31 """Best-effort package kind from PyPI trove classifiers."""
32 text = " ".join(classifiers)
33 if "Environment :: Console" in text or "Topic :: Utilities" in text:
34 return "CLI"
35 return "Library"
38class PyPIAdapter(AdapterBase, RegistryAdapter):
39 @property
40 def registry(self) -> Registry:
41 return Registry.PYPI
43 async def _get_json(self, name: str, client: httpx.AsyncClient) -> dict[str, Any]:
44 resp = await client.get(f"{PYPI_JSON}/{name}/json")
45 resp.raise_for_status()
46 return resp.json()
48 async def fetch_package(self, name: str) -> PackageInfo:
49 async with self._client_scope() as client:
50 data = await self._get_json(name, client)
51 info = data["info"]
53 latest_version = info.get("version", "")
54 versions_data = data.get("releases", {})
55 latest_files = versions_data.get(latest_version, [])
56 upload_time = None
57 if latest_files:
58 upload_time_raw = latest_files[0].get("upload_time", "")
59 if upload_time_raw:
60 with suppress(ValueError, TypeError):
61 upload_time = datetime.fromisoformat(
62 upload_time_raw.replace("Z", "+00:00")
63 )
65 release_files = [
66 ReleaseFile(
67 packagetype=f.get("packagetype", "") or "",
68 size=f.get("size", 0) or 0,
69 filename=f.get("filename", "") or "",
70 )
71 for f in latest_files
72 ]
74 project_urls = info.get("project_urls") or {}
75 homepage = info.get("home_page", "")
76 repo_url = project_urls.get("Source", "")
77 if not repo_url:
78 repo_url = project_urls.get("Repository", "")
79 if not repo_url:
80 repo_url = project_urls.get("Source Code", "")
81 docs_url = info.get("docs_url", "")
82 if not docs_url:
83 docs_url = project_urls.get("Documentation", "")
85 return PackageInfo(
86 name=info["name"],
87 registry=Registry.PYPI,
88 description=info.get("summary", ""),
89 author=info.get("author", ""),
90 license=info.get("license", ""),
91 homepage=homepage,
92 repository_url=repo_url,
93 documentation_url=docs_url,
94 latest_version=latest_version,
95 latest_release_date=upload_time,
96 package_kind=_derive_kind(info.get("classifiers", []) or []),
97 latest_release_files=release_files,
98 )
100 async def fetch_versions(self, name: str) -> list[Version]:
101 async with self._client_scope() as client:
102 data = await self._get_json(name, client)
103 versions_data = data.get("releases", {})
105 versions: list[Version] = []
106 for ver_str, files in versions_data.items():
107 upload_time = None
108 for f in files:
109 raw = f.get("upload_time", "")
110 if raw:
111 try:
112 upload_time = datetime.fromisoformat(
113 raw.replace("Z", "+00:00")
114 )
115 break
116 except (ValueError, TypeError):
117 pass
119 yanked = any(f.get("yanked", False) for f in files)
120 size = sum(f.get("size", 0) or 0 for f in files) or None
122 versions.append(
123 Version(
124 version=ver_str,
125 release_date=upload_time,
126 is_yanked=yanked,
127 size_bytes=size,
128 )
129 )
131 versions.sort(key=lambda v: v.release_date or datetime.min, reverse=True)
132 return versions
134 async def fetch_dependencies(self, name: str, version: str) -> list[Dependency]:
135 async with self._client_scope() as client:
136 data = await self._get_json(name, client)
137 info = data["info"]
138 requires_dist = info.get("requires_dist") or []
140 deps: list[Dependency] = []
141 for raw in requires_dist:
142 if not raw:
143 continue
144 if "extra ==" in raw:
145 continue
146 raw_clean = raw.split(";")[0].strip()
147 if not raw_clean:
148 continue
149 parts = raw_clean.split()
150 dep_name = parts[0].strip()
151 requirement = " ".join(parts[1:]) if len(parts) > 1 else ""
152 deps.append(Dependency(name=dep_name, requirement=requirement))
154 return deps
156 async def fetch_download_trend(
157 self, name: str, days: int = 30
158 ) -> list[DownloadTrendPoint]:
159 async with self._client_scope() as client:
160 try:
161 resp = await client.get(
162 f"{PYPI_STATS}/packages/{name}/overall",
163 params={"mirrors": "false"},
164 )
165 resp.raise_for_status()
166 stats_data = resp.json()
167 except httpx.HTTPError:
168 return []
170 raw_data = stats_data.get("data", [])
171 points = [
172 DownloadTrendPoint(
173 date=entry.get("date", ""),
174 count=entry.get("downloads", 0),
175 )
176 for entry in raw_data
177 ]
178 points.sort(key=lambda p: p.date)
179 return points[-days:] if len(points) > days else points
181 async def fetch_download_counts(self, name: str) -> DownloadCounts:
182 async with self._client_scope() as client:
183 try:
184 resp = await client.get(
185 f"{PYPI_STATS}/packages/{name}/recent",
186 params={"mirrors": "false"},
187 )
188 resp.raise_for_status()
189 data = resp.json()
190 period_data = data.get("data", {})
191 return DownloadCounts(
192 today=period_data.get("last_day", 0),
193 week=period_data.get("last_week", 0),
194 month=period_data.get("last_month", 0),
195 )
196 except httpx.HTTPError:
197 pass
199 # Fallback: compute from trend data
200 trend = await self.fetch_download_trend(name, days=30)
201 if not trend:
202 return DownloadCounts()
203 return DownloadCounts(
204 today=trend[-1].count if trend else 0,
205 week=sum(p.count for p in trend[-7:]),
206 month=sum(p.count for p in trend),
207 )
209 async def fetch_release_notes(self, name: str, version: str) -> str:
210 return "" # fetched via GitHub in utils
212 async def search(self, query: str, limit: int = 10) -> list[SearchResult]:
213 """Search PyPI's public search page (the JSON API has no search route)."""
214 async with self._client_scope() as client:
215 # PyPI's JSON API is dependable for exact package resolution even
216 # when its HTML search page changes markup or is unavailable.
217 try:
218 exact_response = await client.get(f"{PYPI_JSON}/{query}/json")
219 exact_response.raise_for_status()
220 exact_info = exact_response.json().get("info", {})
221 return [
222 SearchResult(
223 name=exact_info.get("name", query),
224 registry=Registry.PYPI,
225 version=exact_info.get("version", ""),
226 description=exact_info.get("summary", "") or "",
227 url=f"https://pypi.org/project/{exact_info.get('name', query)}/",
228 score=1.0,
229 exact=True,
230 )
231 ]
232 except (httpx.HTTPError, ValueError, KeyError) as exc:
233 diagnostics = client.diagnostics
234 if diagnostics is not None:
235 status = (
236 DiagnosticStatus.SUCCESS
237 if isinstance(exc, httpx.HTTPStatusError)
238 and exc.response.status_code == 404
239 else DiagnosticStatus.WARN
240 )
241 message = (
242 "Exact package was not found; trying HTML search"
243 if status is DiagnosticStatus.SUCCESS
244 else "Exact JSON lookup failed; trying HTML search: "
245 + diagnostic_for_http_error(exc)
246 )
247 diagnostics.record(status, "PyPI", message)
248 try:
249 response = await client.get(
250 "https://pypi.org/search/",
251 params={"q": query, "page": 1},
252 headers={"Accept": "text/html"},
253 )
254 response.raise_for_status()
255 except httpx.HTTPError:
256 raise
258 pattern = re.compile(
259 r'data-project-url="([^"]+)"[^>]*>.*?'
260 r'class="package-snippet__name">\s*([^<]+).*?'
261 r'class="package-snippet__version">\s*([^<]+).*?'
262 r'class="package-snippet__description">\s*([^<]*)',
263 re.DOTALL,
264 )
265 results: list[SearchResult] = []
266 for url, name, version, description in pattern.findall(response.text)[:limit]:
267 clean_name = unescape(name).strip()
268 results.append(
269 SearchResult(
270 name=clean_name,
271 registry=Registry.PYPI,
272 version=unescape(version).strip(),
273 description=" ".join(unescape(description).split()),
274 url=f"https://pypi.org{url}" if url.startswith("/") else url,
275 exact=clean_name.lower() == query.lower(),
276 score=1.0 if clean_name.lower() == query.lower() else 0.0,
277 )
278 )
279 return results