Coverage for src/secchi/api/npm.py: 74%
142 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-14 23:28 +0000
1"""npm registry adapter — direct REST API."""
3from __future__ import annotations
5from datetime import UTC, datetime, timedelta
7import httpx
9from secchi.api.base import AdapterBase, RegistryAdapter
10from secchi.models import (
11 Dependency,
12 DownloadCounts,
13 DownloadTrendPoint,
14 PackageInfo,
15 Registry,
16 ReleaseFile,
17 SearchResult,
18 Version,
19)
21NPM_REGISTRY = "https://registry.npmjs.org"
22NPM_DOWNLOADS = "https://api.npmjs.org/downloads"
24# Aware sort sentinel for versions with no parsed release date — release_date
25# is always tz-aware when present, and a naive datetime.min would raise
26# TypeError when compared against it, dropping the whole version list.
27_MIN_DATETIME = datetime.min.replace(tzinfo=UTC)
30def _npm_name(name: str) -> str:
31 """URL-encode an npm package name (handles scoped packages like @scope/name)."""
32 if "/" in name:
33 parts = name.split("/", 1)
34 return f"{parts[0]}/{parts[1]}"
35 return name
38class NpmAdapter(AdapterBase, RegistryAdapter):
39 @property
40 def registry(self) -> Registry:
41 return Registry.NPM
43 async def fetch_package(self, name: str) -> PackageInfo:
44 async with self._client_scope() as client:
45 safe_name = _npm_name(name)
46 resp = await client.get(f"{NPM_REGISTRY}/{safe_name}")
47 resp.raise_for_status()
48 data = resp.json()
50 latest_tag = data.get("dist-tags", {}).get("latest", "")
51 latest_info = (
52 data.get("versions", {}).get(latest_tag, {}) if latest_tag else {}
53 )
55 description = data.get("description", "")
56 author_info = data.get("author", {})
57 author = (
58 author_info.get("name", "")
59 if isinstance(author_info, dict)
60 else str(author_info)
61 if author_info
62 else ""
63 )
65 repo_info = data.get("repository", {})
66 repo_url = (
67 repo_info.get("url", "")
68 if isinstance(repo_info, dict)
69 else str(repo_info)
70 if repo_info
71 else ""
72 )
74 homepage = data.get("homepage", "")
76 if latest_info:
77 latest_date = next(
78 (
79 Version(
80 version=v,
81 release_date=_parse_npm_time(data.get("time", {}).get(v)),
82 )
83 for v in [latest_tag]
84 if v in data.get("time", {})
85 ),
86 Version(version=latest_tag),
87 )
88 latest_release_date = latest_date.release_date
89 else:
90 latest_release_date = None
92 total_downloads = await self._fetch_total_downloads(name)
94 kind = "CLI" if latest_info.get("bin") else "Library"
95 release_files: list[ReleaseFile] = []
96 unpacked = latest_info.get("dist", {}).get("unpackedSize")
97 if unpacked:
98 release_files.append(
99 ReleaseFile(
100 packagetype="npm-package",
101 size=unpacked,
102 filename=f"{name}-{latest_tag}.tgz",
103 )
104 )
106 return PackageInfo(
107 name=data["name"],
108 registry=Registry.NPM,
109 description=description,
110 author=author,
111 license=latest_info.get("license", ""),
112 homepage=homepage,
113 repository_url=repo_url,
114 latest_version=latest_tag,
115 latest_release_date=latest_release_date,
116 total_downloads=total_downloads,
117 package_kind=kind,
118 latest_release_files=release_files,
119 )
121 async def fetch_versions(self, name: str) -> list[Version]:
122 async with self._client_scope() as client:
123 safe_name = _npm_name(name)
124 resp = await client.get(f"{NPM_REGISTRY}/{safe_name}")
125 resp.raise_for_status()
126 data = resp.json()
128 versions: list[Version] = []
129 time_data = data.get("time", {})
130 for ver, info in data.get("versions", {}).items():
131 release_date = _parse_npm_time(time_data.get(ver))
132 size = (
133 info.get("dist", {}).get("unpackedSize")
134 if isinstance(info, dict)
135 else None
136 )
137 versions.append(
138 Version(
139 version=ver,
140 release_date=release_date,
141 size_bytes=size,
142 )
143 )
145 versions.sort(key=lambda v: v.release_date or _MIN_DATETIME, reverse=True)
146 return versions
148 async def fetch_dependencies(self, name: str, version: str) -> list[Dependency]:
149 async with self._client_scope() as client:
150 safe_name = _npm_name(name)
151 resp = await client.get(f"{NPM_REGISTRY}/{safe_name}")
152 resp.raise_for_status()
153 data = resp.json()
155 version_data = data.get("versions", {}).get(version, {})
156 deps: list[Dependency] = []
158 for dep_key, dep_label in [
159 ("dependencies", False),
160 ("devDependencies", True),
161 ("peerDependencies", True),
162 ]:
163 dep_map = version_data.get(dep_key, {})
164 if isinstance(dep_map, dict):
165 for dep_name, req in dep_map.items():
166 deps.append(
167 Dependency(
168 name=dep_name,
169 requirement=str(req),
170 optional=dep_label != "dependencies",
171 )
172 )
174 return deps
176 async def fetch_download_trend(
177 self, name: str, days: int = 30
178 ) -> list[DownloadTrendPoint]:
179 async with self._client_scope() as client:
180 try:
181 if days <= 31:
182 url = f"{NPM_DOWNLOADS}/range/last-month/{name}"
183 else:
184 end = datetime.now(UTC).date()
185 start = end - timedelta(days=days)
186 url = (
187 f"{NPM_DOWNLOADS}/range/{start:%Y-%m-%d}:{end:%Y-%m-%d}/{name}"
188 )
189 resp = await client.get(url)
190 resp.raise_for_status()
191 data = resp.json()
192 except httpx.HTTPError:
193 return []
195 points = [
196 DownloadTrendPoint(date=entry["day"], count=entry["downloads"])
197 for entry in data.get("downloads", [])
198 ]
199 return points[-days:] if len(points) > days else points
201 async def fetch_download_counts(self, name: str) -> DownloadCounts:
202 async with self._client_scope() as client:
203 today = week = month = 0
204 for period, store in [
205 ("last-day", "today"),
206 ("last-week", "week"),
207 ("last-month", "month"),
208 ]:
209 try:
210 resp = await client.get(f"{NPM_DOWNLOADS}/point/{period}/{name}")
211 resp.raise_for_status()
212 val = resp.json().get("downloads", 0)
213 if store == "today":
214 today = val
215 elif store == "week":
216 week = val
217 else:
218 month = val
219 except httpx.HTTPError:
220 pass
221 return DownloadCounts(today=today, week=week, month=month)
223 async def fetch_release_notes(self, name: str, version: str) -> str:
224 async with self._client_scope() as client:
225 safe_name = _npm_name(name)
226 resp = await client.get(f"{NPM_REGISTRY}/{safe_name}")
227 resp.raise_for_status()
228 data = resp.json()
230 version_data = data.get("versions", {}).get(version, {})
231 readme = version_data.get("readme", "")
232 if readme:
233 max_chars = 2000
234 return readme[:max_chars] + ("..." if len(readme) > max_chars else "")
235 return ""
237 async def _fetch_total_downloads(self, name: str) -> int:
238 async with self._client_scope() as client:
239 try:
240 resp = await client.get(f"{NPM_DOWNLOADS}/point/last-year/{name}")
241 resp.raise_for_status()
242 return resp.json().get("downloads", 0)
243 except httpx.HTTPError:
244 return 0
246 async def search(self, query: str, limit: int = 10) -> list[SearchResult]:
247 async with self._client_scope() as client:
248 try:
249 response = await client.get(
250 f"{NPM_REGISTRY}/-/v1/search",
251 params={"text": query, "size": limit},
252 )
253 response.raise_for_status()
254 except httpx.HTTPError:
255 return []
256 results: list[SearchResult] = []
257 for item in response.json().get("objects", [])[:limit]:
258 package = item.get("package", {})
259 name = package.get("name", "")
260 score_data = item.get("score", {})
261 results.append(
262 SearchResult(
263 name=name,
264 registry=Registry.NPM,
265 version=package.get("version", ""),
266 description=package.get("description", "") or "",
267 url=package.get("links", {}).get("npm", ""),
268 score=float(score_data.get("final", 0.0) or 0.0),
269 exact=name.lower() == query.lower(),
270 )
271 )
272 return results
275def _parse_npm_time(raw: str | None) -> datetime | None:
276 if not raw:
277 return None
278 try:
279 return datetime.fromisoformat(raw.replace("Z", "+00:00"))
280 except (ValueError, TypeError):
281 return None