# Copyright (C) 2018-2023 the Software Heritage developers
# License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information
from dataclasses import asdict, dataclass
import logging
from typing import Any, Dict, Iterator, List, Optional
import iso8601
from swh.lister.pattern import CredentialsType, Lister
from swh.scheduler.interface import SchedulerInterface
from swh.scheduler.model import ListedOrigin
logger = logging.getLogger(__name__)
[docs]
@dataclass
class NpmListerState:
"""State of npm lister"""
last_seq: Optional[int] = None
[docs]
class NpmLister(Lister[NpmListerState, List[Dict[str, Any]]]):
"""
List all packages hosted on the npm registry.
The lister is based on the npm replication API powered by a
CouchDB database (https://docs.couchdb.org/en/stable/api/database/).
Args:
scheduler: a scheduler instance
page_size: number of packages info to return per page when querying npm API
incremental: defines if incremental listing should be used, in that case
only modified or new packages since last incremental listing operation
will be returned, otherwise all packages will be listed in lexicographical
order
"""
LISTER_NAME = "npm"
INSTANCE = "npm"
API_BASE_URL = "https://replicate.npmjs.com"
API_INCREMENTAL_LISTING_URL = f"{API_BASE_URL}/_changes"
API_FULL_LISTING_URL = f"{API_BASE_URL}/_all_docs"
PACKAGE_URL_TEMPLATE = "https://www.npmjs.com/package/{package_name}"
def __init__(
self,
scheduler: SchedulerInterface,
url: str = API_FULL_LISTING_URL,
instance: str = INSTANCE,
page_size: int = 1000,
incremental: bool = False,
credentials: CredentialsType = None,
max_origins_per_page: Optional[int] = None,
max_pages: Optional[int] = None,
enable_origins: bool = True,
):
super().__init__(
scheduler=scheduler,
credentials=credentials,
url=url,
instance=instance,
max_origins_per_page=max_origins_per_page,
max_pages=max_pages,
enable_origins=enable_origins,
)
self.page_size = page_size
if not incremental:
# in full listing mode, first package in each page corresponds to the one
# provided as the startkey query parameter value, so we increment the page
# size by one to avoid double package processing
self.page_size += 1
else:
self.url = self.API_INCREMENTAL_LISTING_URL
self.incremental = incremental
self.session.headers.update({"Accept": "application/json"})
[docs]
def state_from_dict(self, d: Dict[str, Any]) -> NpmListerState:
return NpmListerState(**d)
[docs]
def state_to_dict(self, state: NpmListerState) -> Dict[str, Any]:
return asdict(state)
[docs]
def request_params(self, last_package_id: str) -> Dict[str, Any]:
# include package JSON document to get its last update date
params = {"limit": self.page_size, "include_docs": "true"}
if self.incremental:
params["since"] = last_package_id
else:
params["startkey"] = last_package_id
return params
[docs]
def get_pages(self) -> Iterator[List[Dict[str, Any]]]:
last_package_id: str = "0" if self.incremental else '""'
if (
self.incremental
and self.state is not None
and self.state.last_seq is not None
):
last_package_id = str(self.state.last_seq)
while True:
response = self.http_request(
self.url, params=self.request_params(last_package_id)
)
data = response.json()
page = data["results"] if self.incremental else data["rows"]
if not page:
break
if self.incremental or len(page) < self.page_size:
yield page
else:
yield page[:-1]
if len(page) < self.page_size:
break
last_package_id = (
str(page[-1]["seq"]) if self.incremental else f'"{page[-1]["id"]}"'
)
[docs]
def get_origins_from_page(
self, page: List[Dict[str, Any]]
) -> Iterator[ListedOrigin]:
"""Convert a page of Npm repositories into a list of ListedOrigin."""
assert self.lister_obj.id is not None
for package in page:
# no source code to archive here
if not package["doc"].get("versions", {}):
continue
package_name = package["doc"]["name"]
package_latest_version = (
package["doc"].get("dist-tags", {}).get("latest", "")
)
last_update = None
if package_latest_version in package["doc"].get("time", {}):
last_update = iso8601.parse_date(
package["doc"]["time"][package_latest_version]
)
yield ListedOrigin(
lister_id=self.lister_obj.id,
url=self.PACKAGE_URL_TEMPLATE.format(package_name=package_name),
visit_type="npm",
last_update=last_update,
)
[docs]
def commit_page(self, page: List[Dict[str, Any]]):
"""Update the currently stored state using the latest listed page."""
if self.incremental:
last_package = page[-1]
last_seq = last_package["seq"]
if self.state.last_seq is None or last_seq > self.state.last_seq:
self.state.last_seq = last_seq
[docs]
def finalize(self):
if self.incremental and self.state.last_seq is not None:
scheduler_state = self.get_state_from_scheduler()
if (
scheduler_state.last_seq is None
or self.state.last_seq > scheduler_state.last_seq
):
self.updated = True
