Differential D5071 Diff 18128 swh/loader/package/deposit/loader.py

Changeset View

Standalone View

swh/loader/package/deposit/loader.py

	# Copyright (C) 2019-2020 The Software Heritage developers			# Copyright (C) 2019-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

	import datetime			import datetime
	from datetime import timezone			from datetime import timezone
	import json			import json
	import logging			import logging
	from typing import Any, Dict, Iterator, List, Mapping, Optional, Sequence, Tuple, Union			from typing import Any, Dict, Iterator, List, Mapping, Optional, Sequence, Tuple, Union

	import attr			import attr
	import requests			import requests

				from swh.core.config import load_from_envvar
	from swh.loader.package.loader import (			from swh.loader.package.loader import (
	BasePackageInfo,			BasePackageInfo,
	PackageLoader,			PackageLoader,
	RawExtrinsicMetadataCore,			RawExtrinsicMetadataCore,
	)			)
	from swh.loader.package.utils import cached_method, download			from swh.loader.package.utils import cached_method, download
				from swh.loader.pattern import DEFAULT_CONFIG
	from swh.model.hashutil import hash_to_bytes, hash_to_hex			from swh.model.hashutil import hash_to_bytes, hash_to_hex
	from swh.model.model import (			from swh.model.model import (
	MetadataAuthority,			MetadataAuthority,
	MetadataAuthorityType,			MetadataAuthorityType,
	MetadataFetcher,			MetadataFetcher,
	Person,			Person,
	Revision,			Revision,
	RevisionType,			RevisionType,
	Sha1Git,			Sha1Git,
	TimestampWithTimezone,			TimestampWithTimezone,
	)			)
	from swh.storage.algos.snapshot import snapshot_get_all_branches			from swh.storage.algos.snapshot import snapshot_get_all_branches
				from swh.storage.interface import StorageInterface

	logger = logging.getLogger(__name__)			logger = logging.getLogger(__name__)


	def now() -> datetime.datetime:			def now() -> datetime.datetime:
	return datetime.datetime.now(tz=timezone.utc)			return datetime.datetime.now(tz=timezone.utc)


	▲ Show 20 Lines • Show All 61 Lines • ▼ Show 20 Lines

	class DepositLoader(PackageLoader[DepositPackageInfo]):			class DepositLoader(PackageLoader[DepositPackageInfo]):
	"""Load a deposited artifact into swh archive.			"""Load a deposited artifact into swh archive.

	"""			"""

	visit_type = "deposit"			visit_type = "deposit"

	def __init__(self, url: str, deposit_id: str):			def __init__(
				self,
				storage: StorageInterface,
				url: str,
				deposit_id: str,
				deposit_client: "ApiClient",
				max_content_size: Optional[int] = None,
				):
	"""Constructor			"""Constructor

	Args:			Args:
	url: Origin url to associate the artifacts/metadata to			url: Origin url to associate the artifacts/metadata to
	deposit_id: Deposit identity			deposit_id: Deposit identity
				deposit_client: Deposit api client

	"""			"""
	super().__init__(url=url)			super().__init__(storage=storage, url=url, max_content_size=max_content_size)

	config_deposit = self.config["deposit"]
	self.deposit_id = deposit_id			self.deposit_id = deposit_id
	self.client = ApiClient(url=config_deposit["url"], auth=config_deposit["auth"])			self.client = deposit_client

				@classmethod
				def from_configfile(cls, **kwargs: Any):
				"""Instantiate a loader from the configuration loaded from the
				SWH_CONFIG_FILENAME envvar, with potential extra keyword arguments if their
				value is not None.

				Args:
				kwargs: kwargs passed to the loader instantiation

				"""
				config = dict(load_from_envvar(DEFAULT_CONFIG))
				config.update({k: v for k, v in kwargs.items() if v is not None})
				deposit_client = ApiClient(**config.pop("deposit"))
				return cls.from_config(deposit_client=deposit_client, **config)

	def get_versions(self) -> Sequence[str]:			def get_versions(self) -> Sequence[str]:
	# only 1 branch 'HEAD' with no alias since we only have 1 snapshot			# only 1 branch 'HEAD' with no alias since we only have 1 snapshot
	# branch			# branch
	return ["HEAD"]			return ["HEAD"]

	def get_metadata_authority(self) -> MetadataAuthority:			def get_metadata_authority(self) -> MetadataAuthority:
	provider = self.metadata()["provider"]			provider = self.metadata()["provider"]
	▲ Show 20 Lines • Show All 223 Lines • Show Last 20 Lines