Page MenuHomeSoftware Heritage
Differential D5977 Diff 21565 swh/lister/pypi/lister.py

Changeset View

Standalone View

View Options

swh/lister/pypi/lister.py

# Copyright (C) 2018-2021 The Software Heritage developers # Copyright (C) 2018-2021 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution # See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version # License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information # See top-level LICENSE file for more information
from collections import defaultdict
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
import logging import logging
from typing import Iterator, List, Optional from typing import Any, Dict, Iterator, List, Optional, Tuple
from xmlrpc.client import ServerProxy
from bs4 import BeautifulSoup from tenacity.before_sleep import before_sleep_log
import requests
from swh.lister.utils import throttling_retry
from swh.scheduler.interface import SchedulerInterface from swh.scheduler.interface import SchedulerInterface
from swh.scheduler.model import ListedOrigin from swh.scheduler.model import ListedOrigin
from .. import USER_AGENT from ..pattern import CredentialsType, Lister
from ..pattern import CredentialsType, StatelessLister
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
PackageListPage = List[str] # Type returned by listing a page of results
PackageListPage = List[Dict]
# Type returned by the XML-RPC changelog call:
# package, version, release timestamp, description, serial
ChangelogEntry = Tuple[str, str, int, str, int]
# Manipulated package updated type which is a subset information
# of the ChangelogEntry type: package, release timestamp
PackageUpdate = Tuple[str, int]
class PyPILister(StatelessLister[PackageListPage]):
@dataclass
class PyPIListerState:
"""State of PyPI lister"""
anlambertUnsubmitted
Done
Inline Actions

since the pypi instance was visited

anlambert: since the pypi instance was visited
last_serial: Optional[int] = None
"""Last seen serial when visiting the pypi instance"""
class PyPILister(Lister[PyPIListerState, PackageListPage]):
"""List origins from PyPI. """List origins from PyPI.
""" """
LISTER_NAME = "pypi" LISTER_NAME = "pypi"
INSTANCE = "pypi" # As of today only the main pypi.org is used INSTANCE = "pypi" # As of today only the main pypi.org is used
PACKAGE_LIST_URL = "https://pypi.org/pypi" # XML-RPC url
PACKAGE_LIST_URL = "https://pypi.org/simple/"
PACKAGE_URL = "https://pypi.org/project/{package_name}/" PACKAGE_URL = "https://pypi.org/project/{package_name}/"
def __init__( def __init__(
self, self,
scheduler: SchedulerInterface, scheduler: SchedulerInterface,
credentials: Optional[CredentialsType] = None, credentials: Optional[CredentialsType] = None,
): ):
super().__init__( super().__init__(
scheduler=scheduler, scheduler=scheduler,
url=self.PACKAGE_LIST_URL, url=self.PACKAGE_LIST_URL,
instance=self.INSTANCE, instance=self.INSTANCE,
credentials=credentials, credentials=credentials,
) )
self.session = requests.Session() # used as termination condition and if useful, becomes the new state when the
self.session.headers.update( # visit is done
olasdUnsubmitted
Done
Inline Actions

maybe self.last_processed_serial?

olasd: maybe `self.last_processed_serial`?
ardumontAuthorUnsubmitted
Done
Inline Actions

+1

ardumont: +1
{"Accept": "application/html", "User-Agent": USER_AGENT} self.highest_serial: Optional[int] = None
)
def state_from_dict(self, d: Dict[str, Any]) -> PyPIListerState:
return PyPIListerState(last_serial=d.get("last_serial"))
def state_to_dict(self, state: PyPIListerState) -> Dict[str, Any]:
return asdict(state)
@throttling_retry(before_sleep=before_sleep_log(logger, logging.WARNING))
def _changelog_last_serial(self, client: ServerProxy) -> int:
"""Internal detail to allow throttling when calling the changelog last entry"""
serial = client.changelog_last_serial()
assert isinstance(serial, int)
return serial
@throttling_retry(before_sleep=before_sleep_log(logger, logging.WARNING))
def _changelog_since_serial(
anlambertUnsubmitted
Done
Inline Actions

You could use a dataclass PackageUpdate instead of a tuple here for better readability.

anlambert: You could use a dataclass `PackageUpdate` instead of a tuple here for better readability.
ardumontAuthorUnsubmitted
Done
Inline Actions

totally, thx ;)

ardumont: totally, thx ;)
self, client: ServerProxy, serial: int
anlambertUnsubmitted
Done
Inline Actions

Execute the listing of the last updates since last_visit_timestamp

anlambert: Execute the listing of the last updates since last_visit_timestamp
) -> List[ChangelogEntry]:
"""Internal detail to allow throttling when calling the changelog listing"""
ardumontAuthorUnsubmitted
Done
Inline Actions

@olasd to explicit ^ when mocking the ServerProxy, here is the issue with whatever methods i'd like to mock ("the *annoying* implementation detail)

AttributeError: <class 'xmlrpc.client.ServerProxy'> does not have the attribute 'last_serial'
AttributeError: <class 'xmlrpc.client.ServerProxy'> does not have the attribute 'changelog_since_serial'
ardumont: @olasd to explicit ^ when mocking the ServerProxy, here is the issue with whatever methods i'd…
olasdUnsubmitted
Done
Inline Actions

My suggestion would be to substitute xmlrpc.client.ServerProxy() with a (stubbed) class implementing just these two methods with hardcoded return values. I don't know how Mock() or MagicMock() behaves on classes which have dynamically generated attributes/methods, like the ones xmlrpc.client generates.

olasd: My suggestion would be to substitute `xmlrpc.client.ServerProxy()` with a (stubbed) class…
ardumontAuthorUnsubmitted
Done
Inline Actions

I actually don't know how to do what you suggest, i'll check.
Thanks.

ardumont: I actually don't know how to do what you suggest, i'll check. Thanks.
ardumontAuthorUnsubmitted
Done
Inline Actions

reading it more slowly, i think i got it.

ardumont: reading it more slowly, i think i got it.
return client.changelog_since_serial(serial) # type: ignore
def get_pages_from(self, last_serial: Optional[int]) -> Iterator[PackageUpdate]:
"""Execute the listing of the last updates since the last_serial seen. When the
execution is done, this will also set the self.highest_serial attribute so we
anlambertUnsubmitted
Done
Inline Actions

The last visit timestamp

anlambert: The last visit timestamp
can finalize the state of the lister for the next visit.
Note that the indirection method exists to hide the actual rpc calls details so
the testing is actually doable. Technically, the ServerProxy class used does not
expose exactly those methods due to internal implementation detail which makes
the testing hard for no good reason.
def get_pages(self) -> Iterator[PackageListPage]: Args:
last_serial: The last serial seen from a previous visit if any
anlambertUnsubmitted
Done
Inline Actions

Is is not client.changelog instead here ?

anlambert: Is is not `client.changelog` instead here ?
ardumontAuthorUnsubmitted
Done
Inline Actions

totally, here falls apart the mocking part ;)
nice catch!

ardumont: totally, here falls apart the mocking part ;) nice catch!
response = self.session.get(self.PACKAGE_LIST_URL) Yields:
Tuple of (package-name, release-date)
response.raise_for_status() """
client = ServerProxy(self.url)
page = BeautifulSoup(response.content, features="html.parser") highest_serial = self._changelog_last_serial(client)
max_serial = last_serial if last_serial is not None else -1
# Paginate through result of pypi, until we read everything
while max_serial < highest_serial:
for package, _, release_date, _, serial in self._changelog_since_serial(
client, max_serial
):
yield (
package,
release_date,
)
# Compute the max serial so we can stop when done
max_serial = max(max_serial, serial)
self.highest_serial = highest_serial
page_results = [p.text for p in page.find_all("a")] def get_pages(self) -> Iterator[PackageListPage]:
"""Iterate other changelog events per package, determine the max release date for that
package and use that max release date as last_update.
yield page_results """
updated_packages = defaultdict(list)
for package, release_date in self.get_pages_from(self.state.last_serial):
updated_packages[package].append(release_date)
yield [
{
"url": self.PACKAGE_URL.format(package_name=package),
"last_update": datetime.fromtimestamp(max(releases)).replace(
tzinfo=timezone.utc
),
}
for package, releases in updated_packages.items()
]
olasdUnsubmitted
Done
Inline Actions
client = ServerProxy(self.url)
- last_serial = self.state.last_serial
- max_serial = last_serial if last_serial is not None else -1
- highest_serial = self._changelog_last_serial(client)
+ last_processed_serial = -1
+ if self.state.last_serial is not None:
+ last_processed_serial = self.state.last_serial
+ upstream_last_serial = self._changelog_last_serial(client)
# Paginate through result of pypi, until we read everything

I'm a bit confused by these three variables. It seems that last_serial is never reused, so here's my proposal!

olasd: I'm a bit confused by these three variables. It seems that `last_serial` is never reused, so…
ardumontAuthorUnsubmitted
Done
Inline Actions

thx, +1 again

ardumont: thx, +1 again
def get_origins_from_page( def get_origins_from_page(
self, packages_name: PackageListPage self, packages: PackageListPage
) -> Iterator[ListedOrigin]: ) -> Iterator[ListedOrigin]:
"""Convert a page of PyPI repositories into a list of ListedOrigins.""" """Convert a page of PyPI repositories into a list of ListedOrigins."""
assert self.lister_obj.id is not None assert self.lister_obj.id is not None
for package_name in packages_name: for package in packages:
package_url = self.PACKAGE_URL.format(package_name=package_name)
yield ListedOrigin( yield ListedOrigin(
lister_id=self.lister_obj.id, lister_id=self.lister_obj.id,
url=package_url, url=package["url"],
visit_type="pypi", visit_type="pypi",
last_update=None, # available on PyPI JSON API last_update=package["last_update"],
) )
def finalize(self):
"""Finalize the visit state by updating with the new last_serial if updates
actually happened.
"""
self.updated = (
self.state
and self.state.last_serial
and self.state.last_serial < self.highest_serial
olasdUnsubmitted
Done
Inline Actions
highest_serial = self._changelog_last_serial(client)
# Paginate through result of pypi, until we read everything
- while max_serial < highest_serial:
+ while last_processed_serial < last_upstream_serial:
updated_packages = defaultdict(list)
for package, _, release_date, _, serial in self._changelog_since_serial(
client, max_serial
):
updated_packages[package].append(release_date)
# Compute the max serial so we can stop when done
- max_serial = max(max_serial, serial)
+ last_processed_serial = max(last_processed_serial, serial)
# Returns pages of result to flush regularly
yield [
(
pypi_url(package),
datetime.fromtimestamp(max(release_dates)).replace(
tzinfo=timezone.utc
),
)
for package, release_dates in updated_packages.items()
]
- self.highest_serial = highest_serial
+ self.last_processed_serial = last_processed_serial
def get_origins_from_page(
olasd:
) or (not self.state.last_serial and self.highest_serial)
if self.updated:
olasdUnsubmitted
Done
Inline Actions
actually happened.
"""
self.updated = (
self.state
and self.state.last_serial
- and self.state.last_serial < self.highest_serial
- ) or (not self.state.last_serial and self.highest_serial)
+ and self.state.last_serial < self.last_processed_serial
+ ) or (not self.state.last_serial and self.last_processed_serial)
if self.updated:
- self.state.last_serial = self.highest_serial
+ self.state.last_serial = self.last_processed_serial
olasd:
self.state.last_serial = self.highest_serial
About · Developer information · Legal