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