Differential D2582 Diff 9256 swh/web/common/service.py

Changeset View

Standalone View

swh/web/common/service.py

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

import itertools		import itertools
import os		import os
import re		import re

from collections import defaultdict		from collections import defaultdict
from typing import Any, Dict		from typing import Any, Dict, List, Set

from swh.model import hashutil		from swh.model import hashutil

from swh.storage.algos import diff, revisions_walker		from swh.storage.algos import diff, revisions_walker

from swh.model.identifiers import (		from swh.model.identifiers import (
CONTENT, DIRECTORY, RELEASE, REVISION, SNAPSHOT		CONTENT, DIRECTORY, RELEASE, REVISION, SNAPSHOT
)		)
▲ Show 20 Lines • Show All 1,146 Lines • ▼ Show 20 Lines	def lookup_object(object_type: str, object_id: str) -> Dict[str, Any]:
elif object_type == REVISION:		elif object_type == REVISION:
return lookup_revision(object_id)		return lookup_revision(object_id)
elif object_type == SNAPSHOT:		elif object_type == SNAPSHOT:
return lookup_snapshot(object_id)		return lookup_snapshot(object_id)

raise BadInputExc(('Invalid swh object type! Valid types are '		raise BadInputExc(('Invalid swh object type! Valid types are '
f'{CONTENT}, {DIRECTORY}, {RELEASE} '		f'{CONTENT}, {DIRECTORY}, {RELEASE} '
f'{REVISION} or {SNAPSHOT}.'))		f'{REVISION} or {SNAPSHOT}.'))


		vlorentzUnsubmitted Not Done Inline Actions We're trying to use type annotations in new functions, so it should be: def lookup_missing_hashes(grouped_pids: Dict[str, List[bytes]]) -> Set[str]: vlorentz: We're trying to use type annotations in new functions, so it should be: ``` def…
		def lookup_missing_hashes(grouped_pids: Dict[str, List[bytes]]) -> Set[str]:
		"""Lookup missing Software Heritage persistent identifier hash, using
		batch processing.

		Args:
		A dictionary with:
		keys: persistent identifier type
		values: list(bytes) persistent identifier hash
		Returns:
		A set(hexadecimal) of the hashes not found in the storage
		"""
		missing_hashes = []

		for obj_type, obj_ids in grouped_pids.items():
		if obj_type == CONTENT:
		missing_hashes.append(
		storage.content_missing_per_sha1_git(obj_ids))
		elif obj_type == DIRECTORY:
		missing_hashes.append(storage.directory_missing(obj_ids))
		elif obj_type == REVISION:
		missing_hashes.append(storage.revision_missing(obj_ids))
		elif obj_type == RELEASE:
		missing_hashes.append(storage.directory_missing(obj_ids))
		elif obj_type == SNAPSHOT:
		zackUnsubmitted Not Done Inline Actions these should be either a series of if/elif/elif/elif, or a dictionary with a single lookup zack: these should be either a series of if/elif/elif/elif, or a dictionary with a single lookup
		missing_hashes.append(storage.directory_missing(obj_ids))

		missing = set(map(lambda x: hashutil.hash_to_hex(x),
		itertools.chain(*missing_hashes)))

		return missing