diff --git a/swh/web/common/converters.py b/swh/web/common/converters.py --- a/swh/web/common/converters.py +++ b/swh/web/common/converters.py @@ -9,6 +9,8 @@ from swh.core.utils import decode_with_escape from swh.model import hashutil +from swh.storage.interface import PartialBranches + from swh.web.common.typing import OriginInfo, OriginVisitInfo @@ -352,7 +354,7 @@ def from_snapshot(snapshot): - """Convert swh snapshot to serializable snapshot dictionary. + """Convert swh snapshot to serializable (partial) snapshot dictionary. """ sv = from_swh(snapshot, hashess={"id", "target"}, bytess={"next_branch"}) @@ -369,6 +371,22 @@ return sv +def from_partial_branches(branches: PartialBranches): + """Convert PartialBranches to serializable partial snapshot dictionary + + """ + return from_snapshot( + { + "id": branches["id"], + "branches": { + branch_name: branch.to_dict() if branch else None + for (branch_name, branch) in branches["branches"].items() + }, + "next_branch": branches["next_branch"], + } + ) + + def from_directory_entry(dir_entry): """Convert swh directory to serializable directory dictionary. diff --git a/swh/web/common/service.py b/swh/web/common/service.py --- a/swh/web/common/service.py +++ b/swh/web/common/service.py @@ -1046,12 +1046,12 @@ A dict filled with the snapshot content. """ snapshot_id_bin = _to_sha1_bin(snapshot_id) - snapshot = storage.snapshot_get_branches( + partial_branches = storage.snapshot_get_branches( snapshot_id_bin, branches_from.encode(), branches_count, target_types ) - if not snapshot: - raise NotFoundExc("Snapshot with id %s not found!" % snapshot_id) - return converters.from_snapshot(snapshot) + if not partial_branches: + raise NotFoundExc(f"Snapshot with id {snapshot_id} not found!") + return converters.from_partial_branches(partial_branches) def lookup_latest_origin_snapshot( diff --git a/swh/web/tests/conftest.py b/swh/web/tests/conftest.py --- a/swh/web/tests/conftest.py +++ b/swh/web/tests/conftest.py @@ -292,13 +292,13 @@ def snapshot_get_branches( self, snapshot_id, branches_from="", branches_count=1000, target_types=None ): - snp = self.storage.snapshot_get_branches( + partial_branches = self.storage.snapshot_get_branches( hash_to_bytes(snapshot_id), branches_from.encode(), branches_count, target_types, ) - return converters.from_snapshot(snp) + return converters.from_partial_branches(partial_branches) def snapshot_get_head(self, snapshot): if snapshot["branches"]["HEAD"]["target_type"] == "alias":