Source code for swh.lister.conda.lister

# Copyright (C) 2022-2023  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 bz2
from collections import defaultdict
import datetime
import json
import logging
from typing import Any, Dict, Iterator, List, Optional, Tuple

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.
CondaListerPage = Tuple[str, Dict[str, Dict[str, Any]]]

[docs] class CondaLister(StatelessLister[CondaListerPage]): """List Conda ( origins.""" LISTER_NAME = "conda" VISIT_TYPE = "conda" INSTANCE = "conda" BASE_REPO_URL = "" REPO_URL_PATTERN = "{url}/{channel}/{arch}/repodata.json.bz2" ORIGIN_URL_PATTERN = "{channel}/{pkgname}" ARCHIVE_URL_PATTERN = "{url}/{channel}/{arch}/{filename}" def __init__( self, scheduler: SchedulerInterface, credentials: Optional[CredentialsType] = None, url: str = BASE_REPO_URL, instance: str = INSTANCE, channel: str = "", archs: List = [], max_origins_per_page: Optional[int] = None, max_pages: Optional[int] = None, enable_origins: bool = True, ): super().__init__( scheduler=scheduler, credentials=credentials, instance=instance, url=url, max_origins_per_page=max_origins_per_page, max_pages=max_pages, enable_origins=enable_origins, ) str = channel self.archs: List[str] = archs self.packages: Dict[str, Any] = defaultdict(dict) self.package_dates: Dict[str, Any] = defaultdict(list)
[docs] def get_pages(self) -> Iterator[CondaListerPage]: """Yield an iterator which returns 'page'""" for arch in self.archs: repodata_url = self.REPO_URL_PATTERN.format( url=self.url,, arch=arch ) response = self.http_request(url=repodata_url) packages: Dict[str, Any] = json.loads(bz2.decompress(response.content))[ "packages" ] yield (arch, packages)
[docs] def get_origins_from_page(self, page: CondaListerPage) -> Iterator[ListedOrigin]: """Iterate on all pages and yield ListedOrigin instances.""" assert is not None arch, packages = page package_names = set() for filename, package_metadata in packages.items(): package_names.add(package_metadata["name"]) version_key = ( f"{arch}/{package_metadata['version']}-{package_metadata['build']}" ) artifact: Dict[str, Any] = { "filename": filename, "url": self.ARCHIVE_URL_PATTERN.format( url=self.url,, filename=filename, arch=arch, ), "version": version_key, "checksums": {}, } for checksum in ("md5", "sha256"): if checksum in package_metadata: artifact["checksums"][checksum] = package_metadata[checksum] self.packages[package_metadata["name"]][version_key] = artifact package_date = None if "timestamp" in package_metadata: package_date = datetime.datetime.fromtimestamp( package_metadata["timestamp"] / 1e3, datetime.timezone.utc ) elif "date" in package_metadata: package_date = iso8601.parse_date(package_metadata["date"]) if package_date: artifact["date"] = package_date.isoformat() self.package_dates[package_metadata["name"]].append(package_date) for package_name in package_names: package_dates = self.package_dates[package_name] yield ListedOrigin(, visit_type=self.VISIT_TYPE, url=self.ORIGIN_URL_PATTERN.format(, pkgname=package_name ), last_update=max(package_dates, default=None), extra_loader_arguments={ "artifacts": list(self.packages[package_name].values()) }, )