Page MenuHomeSoftware Heritage
Differential D8338 Diff 30824 swh/lister/hackage/lister.py

Changeset View

Standalone View

View Options

swh/lister/hackage/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 logging
from typing import Any, Dict, Iterator, List, Optional
import iso8601
from swh.scheduler.interface import SchedulerInterface
from swh.scheduler.model import ListedOrigin
from ..pattern import CredentialsType, StatelessLister
logger = logging.getLogger(__name__)
# Aliasing the page results returned by `get_pages` method from the lister.
HackageListerPage = List[Dict[str, Any]]
class HackageLister(StatelessLister[HackageListerPage]):
"""List Hackage (The Haskell Package Repository) origins."""
LISTER_NAME = "hackage"
VISIT_TYPE = "hackage"
INSTANCE = "hackage"
anlambertUnsubmitted
Done
Inline Actions

You can remove the user agent setting code, it is now handled in base Lister class.

anlambert: You can remove the user agent setting code, it is now handled in base Lister class.
BASE_URL = "https://hackage.haskell.org/"
PACKAGE_NAMES_URL_PATTERN = "{base_url}packages/search"
PACKAGE_INFO_URL_PATTERN = "{base_url}package/{pkgname}"
def __init__(
self,
scheduler: SchedulerInterface,
credentials: Optional[CredentialsType] = None,
url: Optional[str] = None,
):
super().__init__(
scheduler=scheduler,
credentials=credentials,
instance=self.INSTANCE,
url=url if url else self.BASE_URL,
)
# Ensure to set this with same value as the http api search endpoint use
# (50 as of august 2022)
self.page_size: int = 50
def get_pages(self) -> Iterator[HackageListerPage]:
"""Yield an iterator which returns 'page'
It uses the http api endpoint `https://hackage.haskell.org/packages/search`
to get a list of package names from which we build an origin url.
anlambertUnsubmitted
Done
Inline Actions

you can remove that line, session is now created in base lister class

anlambert: you can remove that line, session is now created in base lister class
Results are paginated.
"""
params = {
"page": 0,
"sortColumn": "default",
"sortDirection": "ascending",
"searchQuery": "(deprecated:any)",
}
data = self.http_request(
url=self.PACKAGE_NAMES_URL_PATTERN.format(base_url=self.url),
method="POST",
json=params,
).json()
nb_entries: int = data["numberOfResults"]
(nb_pages, remainder) = divmod(nb_entries, self.page_size)
if remainder:
nb_pages += 1
yield data["pageContents"]
for page in range(1, nb_pages):
params["page"] = page
data = self.http_request(
url=self.PACKAGE_NAMES_URL_PATTERN.format(base_url=self.url),
method="POST",
anlambertUnsubmitted
Done
Inline Actions

You can remove that method, I added an http_request method in base lister class to deduplicate some code.

anlambert: You can remove that method, I added an `http_request` method in base lister class to…
json=params,
).json()
yield data["pageContents"]
def get_origins_from_page(self, page: HackageListerPage) -> Iterator[ListedOrigin]:
"""Iterate on all pages and yield ListedOrigin instances."""
assert self.lister_obj.id is not None
for entry in page:
pkgname = entry["name"]["display"]
last_update = iso8601.parse_date(entry["lastUpload"])
url = self.PACKAGE_INFO_URL_PATTERN.format(
base_url=self.url, pkgname=pkgname
)
yield ListedOrigin(
lister_id=self.lister_obj.id,
visit_type=self.VISIT_TYPE,
url=url,
last_update=last_update,
anlambertUnsubmitted
Done
Inline Actions

Use this instead:

data = self.http_request(
    url=self.PACKAGE_NAMES_URL_PATTERN.format(base_url=self.url),
    method="POST",
    json=params,
).json()
anlambert: Use this instead: ```lang=python data = self.http_request( url=self.
)
vlorentzUnsubmitted
Done
Inline Actions
# Make a first http api call to get the number of results (quantity of package names)
- first = self.page_request(
+ data = self.page_request(
url=self.PACKAGE_NAMES_URL_PATTERN.format(base_url=self.url), params=params
- )
- entries_qty: int = int(first.json()["numberOfResults"])
-
- pages_qty: int = entries_qty // self.page_size
-
- page: int = 0
+ ).json()
+ nb_entries: int = data["numberOfResults"]
+ (nb_pages, remainder) = divmod(nv_entries, self.page_size)
+ if remainder:
+ nb_pages += 1
+ yield data["pageContents"]
- while pages_qty > 0:
+ for page in range(1, nb_pages):
params["page"] = page
data = self.page_request(
url=self.PACKAGE_NAMES_URL_PATTERN.format(base_url=self.url),
params=params,
).json()
- page += 1
- pages_qty -= 1
yield data["pageContents"]
def get_origins_from_page(self, page: HackageListerPage) -> Iterator[ListedOrigin]:
  1. avoids dropping the last page because // rounds down (using divmod + remainder) (please add a test for that)
  2. avoids fetching the first page twice (by yielding before the loop and starting iteration at page=1)
  3. uses less variables
  4. more pythonic (for loop instead of while, renames *_qty to nb_*)
  5. removes unnecessary type casts
vlorentz: 1. avoids dropping the last page because `//` rounds down (using divmod + remainder) (please…
franckbretAuthorUnsubmitted
Done
Inline Actions

Ok, thanks, better now

franckbret: Ok, thanks, better now
anlambertUnsubmitted
Done
Inline Actions

same as my latest comment above

anlambert: same as my latest comment above
About · Developer information · Legal