Differential D3740 Diff 13183 swh/storage/in_memory.py

Changeset View

Standalone View

swh/storage/in_memory.py

Show First 20 Lines • Show All 49 Lines • ▼ Show 20 Lines	from swh.model.model import (
MetadataAuthorityType,		MetadataAuthorityType,
MetadataFetcher,		MetadataFetcher,
MetadataTargetType,		MetadataTargetType,
RawExtrinsicMetadata,		RawExtrinsicMetadata,
Sha1,		Sha1,
Sha1Git,		Sha1Git,
)		)
from swh.model.hashutil import DEFAULT_ALGORITHMS, hash_to_bytes, hash_to_hex		from swh.model.hashutil import DEFAULT_ALGORITHMS, hash_to_bytes, hash_to_hex
from swh.storage.interface import ListOrder, PagedResult, VISIT_STATUSES		from swh.storage.interface import (
		ListOrder,
		PagedResult,
		PartialBranches,
		VISIT_STATUSES,
		)
from swh.storage.objstorage import ObjStorage		from swh.storage.objstorage import ObjStorage
from swh.storage.utils import now		from swh.storage.utils import now

from .converters import origin_url_to_sha1		from .converters import origin_url_to_sha1
from .exc import StorageArgumentException, HashCollision		from .exc import StorageArgumentException, HashCollision
from .utils import get_partition_bounds_bytes		from .utils import get_partition_bounds_bytes
from .writer import JournalWriter		from .writer import JournalWriter

▲ Show 20 Lines • Show All 521 Lines • ▼ Show 20 Lines	def snapshot_add(self, snapshots: List[Snapshot]) -> Dict:
return {"snapshot:add": count}		return {"snapshot:add": count}

def snapshot_missing(self, snapshots: List[Sha1Git]) -> Iterable[Sha1Git]:		def snapshot_missing(self, snapshots: List[Sha1Git]) -> Iterable[Sha1Git]:
for id in snapshots:		for id in snapshots:
if id not in self._snapshots:		if id not in self._snapshots:
yield id		yield id

def snapshot_get(self, snapshot_id: Sha1Git) -> Optional[Dict[str, Any]]:		def snapshot_get(self, snapshot_id: Sha1Git) -> Optional[Dict[str, Any]]:
return self.snapshot_get_branches(snapshot_id)		d = self.snapshot_get_branches(snapshot_id)
		if d is None:
		return None
		return {
		"id": d["id"],
		"branches": {
		name: branch.to_dict() if branch else None
		for (name, branch) in d["branches"].items()
		},
		"next_branch": d["next_branch"],
		}

def snapshot_get_by_origin_visit(		def snapshot_get_by_origin_visit(
self, origin: str, visit: int		self, origin: str, visit: int
) -> Optional[Dict[str, Any]]:		) -> Optional[Dict[str, Any]]:
origin_url = self._get_origin_url(origin)		origin_url = self._get_origin_url(origin)
if not origin_url:		if not origin_url:
return None		return None

Show All 17 Lines	def snapshot_count_branches(self, snapshot_id: Sha1Git) -> Optional[Dict[str, int]]:
)		)

def snapshot_get_branches(		def snapshot_get_branches(
self,		self,
snapshot_id: Sha1Git,		snapshot_id: Sha1Git,
branches_from: bytes = b"",		branches_from: bytes = b"",
branches_count: int = 1000,		branches_count: int = 1000,
target_types: Optional[List[str]] = None,		target_types: Optional[List[str]] = None,
) -> Optional[Dict[str, Any]]:		) -> Optional[PartialBranches]:
snapshot = self._snapshots.get(snapshot_id)		snapshot = self._snapshots.get(snapshot_id)
if snapshot is None:		if snapshot is None:
return None		return None
sorted_branches = sorted(snapshot.branches.items())		sorted_branches = sorted(snapshot.branches.items())
sorted_branch_names = [k for (k, v) in sorted_branches]		sorted_branch_names = [k for (k, v) in sorted_branches]
from_index = bisect.bisect_left(sorted_branch_names, branches_from)		from_index = bisect.bisect_left(sorted_branch_names, branches_from)
if target_types:		if target_types:
next_branch = None		next_branch = None
Show All 15 Lines	) -> Optional[PartialBranches]:
for (branch_name, branch) in snapshot.branches.items()		for (branch_name, branch) in snapshot.branches.items()
if branch_name in returned_branch_names		if branch_name in returned_branch_names
)		)
if to_index >= len(sorted_branch_names):		if to_index >= len(sorted_branch_names):
next_branch = None		next_branch = None
else:		else:
next_branch = sorted_branch_names[to_index]		next_branch = sorted_branch_names[to_index]

branches = {		return PartialBranches(
name: branch.to_dict() if branch else None		id=snapshot_id, branches=branches, next_branch=next_branch,
for (name, branch) in branches.items()		)
}

return {
"id": snapshot_id,
"branches": branches,
"next_branch": next_branch,
}

def snapshot_get_random(self) -> Sha1Git:		def snapshot_get_random(self) -> Sha1Git:
return random.choice(list(self._snapshots))		return random.choice(list(self._snapshots))

def object_find_by_sha1_git(self, ids: List[Sha1Git]) -> Dict[Sha1Git, List[Dict]]:		def object_find_by_sha1_git(self, ids: List[Sha1Git]) -> Dict[Sha1Git, List[Dict]]:
ret: Dict[Sha1Git, List[Dict]] = {}		ret: Dict[Sha1Git, List[Dict]] = {}
for id_ in ids:		for id_ in ids:
objs = self._objects.get(id_, [])		objs = self._objects.get(id_, [])
▲ Show 20 Lines • Show All 604 Lines • Show Last 20 Lines