Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/package/cran/loader.py
# Copyright (C) 2019-2020 The Software Heritage developers | # Copyright (C) 2019-2020 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 | ||||
import dateutil.parser | import dateutil.parser | ||||
import datetime | import datetime | ||||
from datetime import timezone | |||||
import os | import os | ||||
from os import path | |||||
import logging | import logging | ||||
import re | import re | ||||
from typing import Any, Dict, Iterator, List, Mapping, Optional, Tuple | |||||
from datetime import timezone | import attr | ||||
from os import path | |||||
from typing import Any, Generator, Dict, List, Mapping, Optional, Tuple | |||||
from debian.deb822 import Deb822 | from debian.deb822 import Deb822 | ||||
from swh.loader.package.loader import PackageLoader | from swh.loader.package.loader import BasePackageInfo, PackageLoader | ||||
from swh.loader.package.utils import release_name, artifact_identity | from swh.loader.package.utils import release_name, artifact_identity | ||||
from swh.model.model import ( | from swh.model.model import ( | ||||
Person, | Person, | ||||
TimestampWithTimezone, | TimestampWithTimezone, | ||||
Sha1Git, | Sha1Git, | ||||
Revision, | Revision, | ||||
RevisionType, | RevisionType, | ||||
) | ) | ||||
logger = logging.getLogger(__name__) | logger = logging.getLogger(__name__) | ||||
DATE_PATTERN = re.compile(r"^(?P<year>\d{4})-(?P<month>\d{2})$") | DATE_PATTERN = re.compile(r"^(?P<year>\d{4})-(?P<month>\d{2})$") | ||||
class CRANLoader(PackageLoader): | class CRANPackageInfo(BasePackageInfo): | ||||
raw = attr.ib(type=Dict[str, Any]) | |||||
class CRANLoader(PackageLoader[CRANPackageInfo]): | |||||
visit_type = "cran" | visit_type = "cran" | ||||
def __init__(self, url: str, artifacts: List[Dict]): | def __init__(self, url: str, artifacts: List[Dict]): | ||||
"""Loader constructor. | """Loader constructor. | ||||
Args: | Args: | ||||
url: Origin url to retrieve cran artifact(s) from | url: Origin url to retrieve cran artifact(s) from | ||||
artifacts: List of associated artifact for the origin url | artifacts: List of associated artifact for the origin url | ||||
""" | """ | ||||
super().__init__(url=url) | super().__init__(url=url) | ||||
# explicit what we consider the artifact identity | # explicit what we consider the artifact identity | ||||
self.id_keys = ["url", "version"] | self.id_keys = ["url", "version"] | ||||
self.artifacts = artifacts | self.artifacts = artifacts | ||||
def get_versions(self) -> List[str]: | def get_versions(self) -> List[str]: | ||||
versions = [] | versions = [] | ||||
for artifact in self.artifacts: | for artifact in self.artifacts: | ||||
versions.append(artifact["version"]) | versions.append(artifact["version"]) | ||||
return versions | return versions | ||||
def get_default_version(self) -> str: | def get_default_version(self) -> str: | ||||
return self.artifacts[-1]["version"] | return self.artifacts[-1]["version"] | ||||
def get_package_info( | def get_package_info(self, version: str) -> Iterator[Tuple[str, CRANPackageInfo]]: | ||||
ardumont: might as well make that `Iterator[Tuple[str, CranPackageInfo]]` now. | |||||
self, version: str | |||||
) -> Generator[Tuple[str, Dict[str, Any]], None, None]: | |||||
for a_metadata in self.artifacts: | for a_metadata in self.artifacts: | ||||
url = a_metadata["url"] | url = a_metadata["url"] | ||||
package_version = a_metadata["version"] | package_version = a_metadata["version"] | ||||
if version == package_version: | if version == package_version: | ||||
p_info = { | p_info = CRANPackageInfo( | ||||
"url": url, | url=url, filename=path.basename(url), raw=a_metadata, | ||||
"filename": path.basename(url), | ) | ||||
"raw": a_metadata, | |||||
} | |||||
yield release_name(version), p_info | yield release_name(version), p_info | ||||
def resolve_revision_from( | def resolve_revision_from( | ||||
self, | self, | ||||
known_artifacts: Mapping[bytes, Mapping], | known_artifacts: Mapping[bytes, Mapping], | ||||
artifact_metadata: Mapping[str, Any], | artifact_metadata: Mapping[str, Any], | ||||
) -> Optional[bytes]: | ) -> Optional[bytes]: | ||||
"""Given known_artifacts per revision, try to determine the revision for | """Given known_artifacts per revision, try to determine the revision for | ||||
▲ Show 20 Lines • Show All 115 Lines • Show Last 20 Lines |
might as well make that Iterator[Tuple[str, CranPackageInfo]] now.