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

Changeset View

Standalone View

swh/web/common/service.py

Show First 20 Lines • Show All 1,165 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):
		"""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))
		if obj_type == DIRECTORY:
		missing_hashes.append(storage.directory_missing(obj_ids))
		if obj_type == REVISION:
		missing_hashes.append(storage.revision_missing(obj_ids))
		if obj_type == RELEASE:
		missing_hashes.append(storage.directory_missing(obj_ids))
		if 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