Changeset View
Changeset View
Standalone View
Standalone View
swh/loader/package/archive/loader.py
# Copyright (C) 2019 The Software Heritage developers | # Copyright (C) 2019 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 iso8601 | |||||
import logging | import logging | ||||
from os import path | from os import path | ||||
from typing import Any, Dict, Generator, Mapping, Optional, Sequence, Tuple | from typing import Any, Dict, Iterator, Mapping, Optional, Sequence, Tuple | ||||
import attr | |||||
import iso8601 | |||||
from swh.loader.package.loader import PackageLoader | from swh.loader.package.loader import PackageLoader, BasePackageInfo | ||||
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 ( | ||||
Sha1Git, | Sha1Git, | ||||
Person, | Person, | ||||
TimestampWithTimezone, | TimestampWithTimezone, | ||||
Revision, | Revision, | ||||
RevisionType, | RevisionType, | ||||
) | ) | ||||
logger = logging.getLogger(__name__) | logger = logging.getLogger(__name__) | ||||
SWH_PERSON = Person( | SWH_PERSON = Person( | ||||
name=b"Software Heritage", | name=b"Software Heritage", | ||||
fullname=b"Software Heritage", | fullname=b"Software Heritage", | ||||
email=b"robot@softwareheritage.org", | email=b"robot@softwareheritage.org", | ||||
) | ) | ||||
REVISION_MESSAGE = b"swh-loader-package: synthetic revision message" | REVISION_MESSAGE = b"swh-loader-package: synthetic revision message" | ||||
class ArchiveLoader(PackageLoader): | class ArchivePackageInfo(BasePackageInfo): | ||||
raw = attr.ib(type=Dict[str, Any]) | |||||
class ArchiveLoader(PackageLoader[ArchivePackageInfo]): | |||||
"""Load archive origin's artifact files into swh archive | """Load archive origin's artifact files into swh archive | ||||
""" | """ | ||||
visit_type = "tar" | visit_type = "tar" | ||||
def __init__( | def __init__( | ||||
self, | self, | ||||
Show All 38 Lines | def get_versions(self) -> Sequence[str]: | ||||
return versions | return versions | ||||
def get_default_version(self) -> str: | def get_default_version(self) -> str: | ||||
# It's the most recent, so for this loader, it's the last one | # It's the most recent, so for this loader, it's the last one | ||||
return self.artifacts[-1]["version"] | return self.artifacts[-1]["version"] | ||||
def get_package_info( | def get_package_info( | ||||
self, version: str | self, version: str | ||||
) -> Generator[Tuple[str, Mapping[str, Any]], None, None]: | ) -> Iterator[Tuple[str, ArchivePackageInfo]]: | ||||
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: | ||||
filename = a_metadata.get("filename") | filename = a_metadata.get("filename") | ||||
p_info = { | p_info = ArchivePackageInfo( | ||||
"url": url, | url=url, | ||||
"filename": filename if filename else path.split(url)[-1], | filename=filename if filename else path.split(url)[-1], | ||||
"raw": a_metadata, | raw=a_metadata, | ||||
} | ) | ||||
# FIXME: this code assumes we have only 1 artifact per | # FIXME: this code assumes we have only 1 artifact per | ||||
# versioned package | # versioned package | ||||
yield release_name(version), p_info | yield release_name(version), p_info | ||||
def resolve_revision_from( | def resolve_revision_from( | ||||
self, known_artifacts: Dict, artifact_metadata: Dict | self, known_artifacts: Dict, artifact_metadata: Dict | ||||
) -> Optional[bytes]: | ) -> Optional[bytes]: | ||||
identity = artifact_identity( | identity = artifact_identity( | ||||
Show All 38 Lines |