archive.py
No OneTemporary
Actions

Size

1 KB

Subscribers

None

archive.py
View Options

	from typing import Any, Dict, Iterable, List

	# from functools import lru_cache
	from methodtools import lru_cache

	from swh.storage.interface import StorageInterface


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

	@lru_cache(maxsize=100000)
	def directory_ls(self, id: bytes) -> List[Dict[str, Any]]:
	# TODO: filter unused fields
	return [entry for entry in self.storage.directory_ls(id)]

	def iter_origins(self):
	from swh.storage.algos.origin import iter_origins

	yield from iter_origins(self.storage)

	def iter_origin_visits(self, origin: str):
	from swh.storage.algos.origin import iter_origin_visits

	# TODO: filter unused fields
	yield from iter_origin_visits(self.storage, origin)

	def iter_origin_visit_statuses(self, origin: str, visit: int):
	from swh.storage.algos.origin import iter_origin_visit_statuses

	# TODO: filter unused fields
	yield from iter_origin_visit_statuses(self.storage, origin, visit)

	def release_get(self, ids: Iterable[bytes]):
	# TODO: filter unused fields
	yield from self.storage.release_get(list(ids))

	def revision_get(self, ids: Iterable[bytes]):
	# TODO: filter unused fields
	yield from self.storage.revision_get(list(ids))

	def snapshot_get_all_branches(self, snapshot: bytes):
	from swh.storage.algos.snapshot import snapshot_get_all_branches

	# TODO: filter unused fields
	return snapshot_get_all_branches(self.storage, snapshot)