Changeset View
Changeset View
Standalone View
Standalone View
swh/storage/in_memory.py
Show First 20 Lines • Show All 577 Lines • ▼ Show 20 Lines | def snapshot_get_by_origin_visit(self, origin, visit): | ||||
visit = self._origin_visit_get_updated(origin_url, visit) | visit = self._origin_visit_get_updated(origin_url, visit) | ||||
snapshot_id = visit.snapshot | snapshot_id = visit.snapshot | ||||
if snapshot_id: | if snapshot_id: | ||||
return self.snapshot_get(snapshot_id) | return self.snapshot_get(snapshot_id) | ||||
else: | else: | ||||
return None | return None | ||||
def snapshot_get_latest(self, origin, allowed_statuses=None): | |||||
origin_url = self._get_origin_url(origin) | |||||
if not origin_url: | |||||
return | |||||
visit = self.origin_visit_get_latest( | |||||
origin_url, allowed_statuses=allowed_statuses, require_snapshot=True | |||||
) | |||||
if visit and visit["snapshot"]: | |||||
snapshot = self.snapshot_get(visit["snapshot"]) | |||||
if not snapshot: | |||||
raise StorageArgumentException( | |||||
"last origin visit references an unknown snapshot" | |||||
) | |||||
return snapshot | |||||
def snapshot_count_branches(self, snapshot_id): | def snapshot_count_branches(self, snapshot_id): | ||||
snapshot = self._snapshots[snapshot_id] | snapshot = self._snapshots[snapshot_id] | ||||
return collections.Counter( | return collections.Counter( | ||||
branch.target_type.value if branch else None | branch.target_type.value if branch else None | ||||
for branch in snapshot.branches.values() | for branch in snapshot.branches.values() | ||||
) | ) | ||||
def snapshot_get_branches( | def snapshot_get_branches( | ||||
▲ Show 20 Lines • Show All 602 Lines • Show Last 20 Lines |