Differential D6256 Diff 22653 swh/provenance/storage/archive.py

Changeset View

Standalone View

swh/provenance/storage/archive.py

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

	from datetime import datetime			from datetime import datetime
	from typing import Any, Dict, Iterable, Set, Tuple			from typing import Any, Dict, Iterable, Set, Tuple

				from swh.core.statsd import statsd
	from swh.model.model import ObjectType, Sha1Git, TargetType			from swh.model.model import ObjectType, Sha1Git, TargetType
	from swh.storage.interface import StorageInterface			from swh.storage.interface import StorageInterface


	class ArchiveStorage:			class ArchiveStorage:
	def __init__(self, storage: StorageInterface) -> None:			def __init__(self, storage: StorageInterface) -> None:
	self.storage = storage			self.storage = storage

				@statsd.timed(
				metric="swh_provenance_archive_api_accesstime_seconds",
				tags={"method": "directory_ls"},
				)
	def directory_ls(self, id: Sha1Git) -> Iterable[Dict[str, Any]]:			def directory_ls(self, id: Sha1Git) -> Iterable[Dict[str, Any]]:
	# TODO: add file size filtering			# TODO: add file size filtering
	for entry in self.storage.directory_ls(id):			for entry in self.storage.directory_ls(id):
	yield {			yield {
	"name": entry["name"],			"name": entry["name"],
	"target": entry["target"],			"target": entry["target"],
	"type": entry["type"],			"type": entry["type"],
	}			}

				@statsd.timed(
				metric="swh_provenance_archive_api_accesstime_seconds",
				tags={"method": "revision_get_parents"},
				)
	def revision_get_parents(self, id: Sha1Git) -> Iterable[Sha1Git]:			def revision_get_parents(self, id: Sha1Git) -> Iterable[Sha1Git]:
	rev = self.storage.revision_get([id])[0]			rev = self.storage.revision_get([id])[0]
	if rev is not None:			if rev is not None:
	yield from rev.parents			yield from rev.parents

				@statsd.timed(
				metric="swh_provenance_archive_api_accesstime_seconds",
				tags={"method": "snapshot_get_heads"},
				)
	def snapshot_get_heads(self, id: Sha1Git) -> Iterable[Sha1Git]:			def snapshot_get_heads(self, id: Sha1Git) -> Iterable[Sha1Git]:
	from swh.core.utils import grouper			from swh.core.utils import grouper
	from swh.storage.algos.snapshot import snapshot_get_all_branches			from swh.storage.algos.snapshot import snapshot_get_all_branches

	snapshot = snapshot_get_all_branches(self.storage, id)			snapshot = snapshot_get_all_branches(self.storage, id)
	assert snapshot is not None			assert snapshot is not None

	targets_set = set()			targets_set = set()
	Show All 25 Lines