Differential D8287 Diff 29921 swh/lister/pubdev/lister.py

Changeset View

Standalone View

swh/lister/pubdev/lister.py

This file was added.

				# Copyright (C) 2022 The Software Heritage developers
				# See the AUTHORS file at the top-level directory of this distribution
				# License: GNU General Public License version 3, or any later version
				# See top-level LICENSE file for more information
				import datetime
				import logging
				from typing import Any, Dict, Iterator, List, Optional

				import requests
				from tenacity.before_sleep import before_sleep_log

				from swh.lister.utils import throttling_retry
				from swh.scheduler.interface import SchedulerInterface
				from swh.scheduler.model import ListedOrigin

				from .. import USER_AGENT
				from ..pattern import CredentialsType, StatelessLister

				logger = logging.getLogger(__name__)

				# Aliasing the page results returned by `get_pages` method from the lister.
				PubDevListerPage = Dict[str, Any]


				class PubDevLister(StatelessLister[PubDevListerPage]):
				"""List pub.dev (Dart, Flutter) origins."""

				LISTER_NAME = "pubdev"
				VISIT_TYPE = "pubdev"
				INSTANCE = "pubdev"

				BASE_URL = "https://pub.dev/api/"
				PACKAGE_NAMES_URL_PATTERN = "{base_url}package-names"
				PACKAGE_INFO_URL_PATTERN = "{base_url}packages/{pkgname}"

				def __init__(
				self,
				scheduler: SchedulerInterface,
				credentials: Optional[CredentialsType] = None,
				):
				super().__init__(
				scheduler=scheduler,
				credentials=credentials,
				instance=self.INSTANCE,
				url=self.BASE_URL,
				)
				self.session = requests.Session()
				self.session.headers.update(
				{
				"Accept": "application/json",
				"User-Agent": USER_AGENT,
				}
				)

				@throttling_retry(before_sleep=before_sleep_log(logger, logging.WARNING))
				def page_request(self, url: str, params: Dict[str, Any]) -> requests.Response:

				logger.info("Fetching URL %s with params %s", url, params)

				response = self.session.get(url, params=params)
				if response.status_code != 200:
				logger.warning(
				"Unexpected HTTP status code %s on %s: %s",
				response.status_code,
				response.url,
				response.content,
				)
				response.raise_for_status()

				return response

				def get_pages(self) -> Iterator[PubDevListerPage]:
				"""Yield an iterator which returns 'page'

				It uses the api provided by https://pub.dev/api/ to find Dart and Flutter package
				origins.

				The first http api call get "{BASE_URL}package-names" to retrieve a sorted list
				of all package names.

				Each page will corresponds to a package name and an url("{BASE_URL}packages/{pkgname}")
				from which another http api call is made to get package metadata (versions, archive
				url, last update)
				"""
				response = self.page_request(
				url=self.PACKAGE_NAMES_URL_PATTERN.format(base_url=self.url), params={}
				)
				pkgnames = response.json()["packages"]
				for pkgname in pkgnames:
				vlorentzUnsubmitted Done Inline Actions I feel it would make more sense to move this loop to `get_origins_from_page`, since we only have one page. (Not a big deal, in the end the result is the same) vlorentz: I feel it would make more sense to move this loop to `get_origins_from_page`, since we only…
				origin_url: str = self.PACKAGE_INFO_URL_PATTERN.format(
				base_url=self.url, pkgname=pkgname
				)
				res = self.page_request(
				url=origin_url,
				params={},
				)
				page = res.json()
				page["url"] = origin_url
				yield page

				def get_origins_from_page(self, page: PubDevListerPage) -> Iterator[ListedOrigin]:
				"""Iterate on all pages and yield ListedOrigin instances.

				It uses the `packages/{pkgname}` http api endpoint as an origin and adds `artifacts`
				and `pubdev_metadata` entries to 'extra_loader_arguments'.
				vlorentzUnsubmitted Done Inline Actions `last_update` should be `None` when the lister doesn't know when the package was last updated. Setting a non-`None` value would make the scheduler believe we know for sure there is an update, so it would prioritize loading this origin over others. vlorentz: `last_update` should be `None` when the lister doesn't know when the package was last updated.

				`artifacts` describe the archive file to download and `pubdev_metadata` store some
				metadata that can be useful for the loader.
				"""
				assert self.lister_obj.id is not None

				url: str = self.PACKAGE_INFO_URL_PATTERN.format(
				base_url=self.url, pkgname=page["name"]
				)
				latest: Dict[str, Any] = page["latest"]
				versions: List[Dict[str, Any]] = page["versions"]

				last_update = datetime.datetime.fromisoformat(
				latest["published"].replace("Z", "+00:00")
				)

				artifacts: List[Dict[str, Any]] = []
				for version in versions:
				artifact = {
				"version": version["version"],
				"url": version["archive_url"],
				}
				artifacts.append(artifact)

				pubdev_metadata: List[Dict[str, Any]] = versions

				yield ListedOrigin(
				lister_id=self.lister_obj.id,
				visit_type=self.VISIT_TYPE,
				url=url,
				last_update=last_update,
				extra_loader_arguments={
				"artifacts": artifacts,
				"pubdev_metadata": pubdev_metadata,
				},
				)