Page MenuHomeSoftware Heritage
Differential D8033 Diff 29383 swh/lister/aur/lister.py

Changeset View

Standalone View

View Options

swh/lister/aur/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 gzip
import json
import logging
from pathlib import Path
from typing import Any, Dict, Iterator, Optional
import requests
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.
AurListerPage = Dict[str, Any]
ardumontUnsubmitted
Done
Inline Actions
# Aliasing the page results returned by `get_pages` method from the lister.
- AurListerPage = List[Dict[str, Any]]
+ AurListerPage = 'Dict[str, Any]
class AurLister(StatelessLister[AurListerPage]):

No need if you always return one element?

ardumont: No need if you always return one element?
ardumontUnsubmitted
Done
Inline Actions

i mean, your code below returns page of one element if i read it correctly.

ardumont: i mean, your code below returns page of one element if i read it correctly.
franckbretAuthorUnsubmitted
Done
Inline Actions

yep

franckbret: yep
class AurLister(StatelessLister[AurListerPage]):
"""List Arch User Repository (AUR) origins.
Given an url (used as a base url, default is 'https://aur.archlinux.org'),
download a 'packages-meta-v1.json.gz' which contains a json file listing all
existing packages definitions.
Each entry describes the latest released version of a package. The origin url
ardumontUnsubmitted
Done
Inline Actions
existing packages definitions.
- Each entry describe the latest released version of a package. The origin url
+ Each entry describes the latest released version of a package. The origin url
for a package is built using 'pkgname' and corresponds to a git repository.
ardumont:
for a package is built using 'pkgname' and corresponds to a git repository.
An rpc api exists but it is recommended to save bandwidth so it's not used. See
ardumontUnsubmitted
Done
Inline Actions
for a package is built using 'pkgname' and corresponds to a git repository.
- An rpc api exists, but its the recommended way to save bandwidth. See
+ An rpc api exists but it is recommended to save bandwidth so it's not used. See
https://lists.archlinux.org/pipermail/aur-general/2021-November/036659.html
ardumont:
https://lists.archlinux.org/pipermail/aur-general/2021-November/036659.html
for more on this.
"""
LISTER_NAME = "aur"
VISIT_TYPE = "aur"
INSTANCE = "aur"
BASE_URL = "https://aur.archlinux.org"
DEFAULT_PACKAGES_INDEX_URL = "{base_url}/packages-meta-v1.json.gz"
PACKAGE_VCS_URL_PATTERN = "{base_url}/{pkgname}.git"
PACKAGE_SNAPSHOT_URL_PATTERN = "{base_url}/cgit/aur.git/snapshot/{pkgname}.tar.gz"
DESTINATION_PATH = Path("/tmp/aur_archive")
ardumontUnsubmitted
Done
Inline Actions

I gather that's a working directory. If so, you'll need to have some post-listing routine cleanup.

ardumont: I gather that's a working directory. If so, you'll need to have some post-listing routine…
franckbretAuthorUnsubmitted
Done
Inline Actions

Yes It is. Have you an example of post-listing routine action?
Same pattern used for arch and crates, I guess I need to do cleanup to

franckbret: Yes It is. Have you an example of post-listing routine action? Same pattern used for arch and…
franckbretAuthorUnsubmitted
Done
Inline Actions

Is finalize() suitable for that purpose?

franckbret: Is finalize() suitable for that purpose?
vlorentzUnsubmitted
Done
Inline Actions

yes

vlorentz: yes
def __init__(
self,
scheduler: SchedulerInterface,
credentials: Optional[CredentialsType] = None,
):
super().__init__(
scheduler=scheduler,
credentials=credentials,
instance=self.INSTANCE,
url=self.BASE_URL,
)
def download_index_archive(self) -> Path:
"""Build an url based on self.DEFAULT_PACKAGES_INDEX_URL format string,
and download the archive to self.DESTINATION_PATH
Returns:
a directory Path where the archive has been downloaded to.
"""
url = self.DEFAULT_PACKAGES_INDEX_URL.format(base_url=self.url)
filename = url.split("/")[-1]
destination = Path(self.DESTINATION_PATH, filename)
if not Path(self.DESTINATION_PATH).exists():
Path(self.DESTINATION_PATH).mkdir()
response = requests.get(url, stream=True)
destination.write_bytes(response.raw.read())
assert destination.exists()
return destination
def get_pages(self) -> Iterator[AurListerPage]:
"""Yield an iterator which returns 'page'
Each page corresponds to a package with a 'version', an 'url' for a Git
repository, a 'project_url' which represents the upstream project url and
a canonical 'snapshot_url' from which a tar.gz archive of the package can
be downloaded.
"""
index = self.download_index_archive()
with gzip.open(index, "rb") as f:
assert f.readable()
file_content = f.read()
packages = json.loads(file_content)
assert packages
for package in packages:
pkgname = package["Name"]
version = package["Version"]
project_url = package["URL"]
last_modified = datetime.datetime.fromtimestamp(
float(package["LastModified"]), tz=datetime.timezone.utc
).isoformat()
yield {
"pkgname": pkgname,
"version": version,
"url": self.PACKAGE_VCS_URL_PATTERN.format(
base_url=self.BASE_URL, pkgname=pkgname
),
"snapshot_url": self.PACKAGE_SNAPSHOT_URL_PATTERN.format(
base_url=self.BASE_URL, pkgname=pkgname
),
"project_url": project_url,
"last_modified": last_modified,
}
def get_origins_from_page(self, origin: AurListerPage) -> Iterator[ListedOrigin]:
"""Iterate on all pages and yield ListedOrigin instances.
It uses the vcs (Git) url as an origin and adds other information to an
ardumontUnsubmitted
Done
Inline Actions
).isoformat()
- page = [
- {
+ yield {
"pkgname": pkgname,
"version": version,
"url": self.PACKAGE_VCS_URL_PATTERN.format(
base_url=self.BASE_URL, pkgname=pkgname
),
"snapshot_url": self.PACKAGE_SNAPSHOT_URL_PATTERN.format(
base_url=self.BASE_URL, pkgname=pkgname
),
"project_url": project_url,
"last_modified": last_modified,
}
- ]
-
- yield page
def get_origins_from_page(self, page: AurListerPage) -> Iterator[ListedOrigin]:

I've also adapted according to my type suggestion early on.

ardumont: I've also adapted according to my type suggestion early on.
'artifacts' entry of 'extra_loader_arguments'
"""
assert self.lister_obj.id is not None
yield ListedOrigin(
lister_id=self.lister_obj.id,
visit_type=self.VISIT_TYPE,
url=origin["url"],
last_update=datetime.datetime.fromisoformat(origin["last_modified"]),
extra_loader_arguments={
"artifacts": [origin],
},
)
ardumontUnsubmitted
Done
Inline Actions
yield page
- def get_origins_from_page(self, page: AurListerPage) -> Iterator[ListedOrigin]:
+ def get_origins_from_page(self, origin: AurListerPage) -> Iterator[ListedOrigin]:
"""Iterate on all pages and yield ListedOrigin instances.
It uses the vcs (Git) url as an origin and adds other information to an
'artifacts' entry of 'extra_loader_arguments'
"""
assert self.lister_obj.id is not None
- for origin in page:
- yield ListedOrigin(
- lister_id=self.lister_obj.id,
- visit_type=self.VISIT_TYPE,
- url=origin["url"],
- last_update=datetime.datetime.fromisoformat(origin["last_modified"]),
- extra_loader_arguments={
- "artifacts": [origin],
- },
- )
+ yield ListedOrigin(
+ lister_id=self.lister_obj.id,
+ visit_type=self.VISIT_TYPE,
+ url=origin["url"],
+ last_update=datetime.datetime.fromisoformat(origin["last_modified"]),
+ extra_loader_arguments={
+ "artifacts": [origin],
+ },
+ )

in effect, you only read one origin per "page".

ardumont: in effect, you only read one origin per "page".
About · Developer information · Legal