Differential D5181 Diff 18532 swh/web/common/archive.py

Changeset View

Standalone View

swh/web/common/archive.py

# Copyright (C) 2015-2021 The Software Heritage developers		# Copyright (C) 2015-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 Affero General Public License version 3, or any later version		# License: GNU Affero 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

from collections import defaultdict		from collections import defaultdict
import itertools		import itertools
import os		import os
import re		import re
from typing import Any, Dict, Iterable, Iterator, List, Optional, Set, Tuple, Union		from typing import Any, Dict, Iterable, Iterator, List, Optional, Set, Tuple, Union
from urllib.parse import urlparse		from urllib.parse import urlparse

from swh.model import hashutil		from swh.model import hashutil
from swh.model.identifiers import CONTENT, DIRECTORY, RELEASE, REVISION, SNAPSHOT		from swh.model.identifiers import CONTENT, DIRECTORY, RELEASE, REVISION, SNAPSHOT
from swh.model.model import OriginVisit, Revision		from swh.model.model import (
		MetadataAuthority,
		MetadataAuthorityType,
		OriginVisit,
		Revision,
		)
from swh.storage.algos import diff, revisions_walker		from swh.storage.algos import diff, revisions_walker
from swh.storage.algos.origin import origin_get_latest_visit_status		from swh.storage.algos.origin import origin_get_latest_visit_status
from swh.storage.algos.snapshot import snapshot_get_latest, snapshot_resolve_alias		from swh.storage.algos.snapshot import snapshot_get_latest, snapshot_resolve_alias
from swh.vault.exc import NotFoundExc as VaultNotFoundExc		from swh.vault.exc import NotFoundExc as VaultNotFoundExc
from swh.web import config		from swh.web import config
from swh.web.common import converters, query		from swh.web.common import converters, query
from swh.web.common.exc import BadInputExc, NotFoundExc		from swh.web.common.exc import BadInputExc, NotFoundExc
from swh.web.common.typing import (		from swh.web.common.typing import (
▲ Show 20 Lines • Show All 385 Lines • ▼ Show 20 Lines	def lookup_origin_intrinsic_metadata(origin_url: str) -> Dict[str, Any]:

match = _first_element(idx_storage.origin_intrinsic_metadata_get(origins))		match = _first_element(idx_storage.origin_intrinsic_metadata_get(origins))
result = {}		result = {}
if match:		if match:
result = match.metadata		result = match.metadata
return result		return result


		def lookup_origin_raw_extrinsic_metadata(origin_url: str) -> Dict[str, Any]:
		"""Return extrinsic metadata for origin whose origin matches given origin.

		Args:
		origin_url: origin url

		Raises:
		NotFoundExc when the origin is not found

		Returns:
		origin metadata.

		"""
		origins = [origin_url]
		origin_info = storage.origin_get(origins)[0]
		# assert 0 -> for using pytest --pdb
		if not origin_info:
		raise NotFoundExc(f"Origin with url {origin_url} not found!")
		SWH_AUTHORITY = MetadataAuthority(
		type=MetadataAuthorityType.REGISTRY, url="https://softwareheritage.org/"
		) # Fixing to SWH_AUTHORITY for now
		match = _first_element(
		storage.raw_extrinsic_metadata_get(
		"origin", origin_url, authority=SWH_AUTHORITY
		)
		)
		return match


def _to_sha1_bin(sha1_hex):		def _to_sha1_bin(sha1_hex):
_, sha1_git_bin = query.parse_hash_with_algorithms_or_throws(		_, sha1_git_bin = query.parse_hash_with_algorithms_or_throws(
sha1_hex, ["sha1"], "Only sha1_git is supported." # HACK: sha1_git really		sha1_hex, ["sha1"], "Only sha1_git is supported." # HACK: sha1_git really
)		)
return sha1_git_bin		return sha1_git_bin


def _check_directory_exists(sha1_git, sha1_git_bin):		def _check_directory_exists(sha1_git, sha1_git_bin):
▲ Show 20 Lines • Show All 970 Lines • Show Last 20 Lines