Changeset View
Changeset View
Standalone View
Standalone View
swh/web/browse/snapshot_context.py
| # Copyright (C) 2018-2022 The Software Heritage developers | # Copyright (C) 2018-2022 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 Affero General Public License version 3, or any later version | # License: GNU Affero 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 | ||||
| # Utility module for browsing the archive in a snapshot context. | # Utility module for browsing the archive in a snapshot context. | ||||
| from collections import defaultdict | from collections import defaultdict | ||||
| from typing import Any, Dict, List, Optional, Tuple | from typing import Any, Dict, List, Optional, Tuple | ||||
| from django.http import HttpRequest, HttpResponse | from django.http import HttpRequest, HttpResponse | ||||
| from django.shortcuts import render | from django.shortcuts import redirect, render | ||||
| from django.utils.html import escape | from django.utils.html import escape | ||||
| from swh.model.hashutil import hash_to_bytes | from swh.model.hashutil import hash_to_bytes | ||||
| from swh.model.model import Snapshot | from swh.model.model import Snapshot | ||||
| from swh.model.swhids import CoreSWHID, ObjectType | from swh.model.swhids import CoreSWHID, ObjectType | ||||
| from swh.web.browse.utils import ( | from swh.web.browse.utils import ( | ||||
| format_log_entries, | format_log_entries, | ||||
| gen_release_link, | gen_release_link, | ||||
| Show All 39 Lines | ) -> Optional[SnapshotBranchInfo]: | ||||
| if filtered_branches: | if filtered_branches: | ||||
| return filtered_branches[0] | return filtered_branches[0] | ||||
| else: | else: | ||||
| # case where a large branches list has been truncated | # case where a large branches list has been truncated | ||||
| snp = archive.lookup_snapshot( | snp = archive.lookup_snapshot( | ||||
| snapshot_id, | snapshot_id, | ||||
| branches_from=branch_name, | branches_from=branch_name, | ||||
| branches_count=1, | branches_count=1, | ||||
| target_types=["revision", "alias"], | target_types=["revision", "alias", "content", "directory"], | ||||
| # pull request branches must be browsable even if they are hidden | # pull request branches must be browsable even if they are hidden | ||||
| # by default in branches list | # by default in branches list | ||||
| branch_name_exclude_prefix=None, | branch_name_exclude_prefix=None, | ||||
| ) | ) | ||||
| snp_branch, _, _ = process_snapshot_branches(snp) | snp_branch, _, _ = process_snapshot_branches(snp) | ||||
| if snp_branch and snp_branch[0]["name"] == branch_name: | if snp_branch and snp_branch[0]["name"] == branch_name: | ||||
| branches.append(snp_branch[0]) | branches.append(snp_branch[0]) | ||||
| return snp_branch[0] | return snp_branch[0] | ||||
| ▲ Show 20 Lines • Show All 45 Lines • ▼ Show 20 Lines | |||||
| ) -> None: | ) -> None: | ||||
| """ | """ | ||||
| Utility function to raise an exception when a specified branch/release | Utility function to raise an exception when a specified branch/release | ||||
| can not be found. | can not be found. | ||||
| """ | """ | ||||
| if branch_type == "branch": | if branch_type == "branch": | ||||
| branch_type = "Branch" | branch_type = "Branch" | ||||
| branch_type_plural = "branches" | branch_type_plural = "branches" | ||||
| target_type = "revision" | target_type = "branch" | ||||
| else: | else: | ||||
| branch_type = "Release" | branch_type = "Release" | ||||
| branch_type_plural = "releases" | branch_type_plural = "releases" | ||||
| target_type = "release" | target_type = "release" | ||||
| if snapshot_id and snapshot_sizes[target_type] == 0: | if snapshot_id and snapshot_sizes[target_type] == 0: | ||||
| msg = "Snapshot with id %s has an empty list" " of %s!" % ( | msg = "Snapshot with id %s has an empty list" " of %s!" % ( | ||||
| snapshot_id, | snapshot_id, | ||||
| ▲ Show 20 Lines • Show All 58 Lines • ▼ Show 20 Lines | ) -> Tuple[List[SnapshotBranchInfo], List[SnapshotReleaseInfo], Dict[str, Any]]: | ||||
| revision_to_release = defaultdict(set) | revision_to_release = defaultdict(set) | ||||
| release_to_branch = defaultdict(set) | release_to_branch = defaultdict(set) | ||||
| for branch_name, target in snapshot_branches.items(): | for branch_name, target in snapshot_branches.items(): | ||||
| if not target: | if not target: | ||||
| # FIXME: display branches with an unknown target anyway | # FIXME: display branches with an unknown target anyway | ||||
| continue | continue | ||||
| target_id = target["target"] | target_id = target["target"] | ||||
| target_type = target["target_type"] | target_type = target["target_type"] | ||||
| if target_type == "revision": | if target_type in ("content", "directory", "revision"): | ||||
| branches[branch_name] = SnapshotBranchInfo( | branches[branch_name] = SnapshotBranchInfo( | ||||
| name=branch_name, | name=branch_name, | ||||
| alias=False, | alias=False, | ||||
| revision=target_id, | target_type=target_type, | ||||
| target=target_id, | |||||
| date=None, | date=None, | ||||
| directory=None, | directory=None, | ||||
| message=None, | message=None, | ||||
| url=None, | url=None, | ||||
| ) | ) | ||||
| if target_type == "revision": | |||||
| revision_to_branch[target_id].add(branch_name) | revision_to_branch[target_id].add(branch_name) | ||||
| elif target_type == "release": | elif target_type == "release": | ||||
| release_to_branch[target_id].add(branch_name) | release_to_branch[target_id].add(branch_name) | ||||
| elif target_type == "alias": | elif target_type == "alias": | ||||
| branch_aliases[branch_name] = target_id | branch_aliases[branch_name] = target_id | ||||
| # FIXME: handle pointers to other object types | # FIXME: handle pointers to other object types | ||||
| def _add_release_info(branch, release, alias=False): | def _add_release_info(branch, release, alias=False): | ||||
| releases[branch] = SnapshotReleaseInfo( | releases[branch] = SnapshotReleaseInfo( | ||||
| name=release["name"], | name=release["name"], | ||||
| alias=alias, | alias=alias, | ||||
| branch_name=branch, | branch_name=branch, | ||||
| date=format_utc_iso_date(release["date"]), | date=format_utc_iso_date(release["date"]), | ||||
| directory=None, | directory=None, | ||||
| id=release["id"], | id=release["id"], | ||||
| message=release["message"], | message=release["message"], | ||||
| target_type=release["target_type"], | target_type=release["target_type"], | ||||
| target=release["target"], | target=release["target"], | ||||
| url=None, | url=None, | ||||
| ) | ) | ||||
| def _add_branch_info(branch, revision, alias=False): | def _add_branch_info(branch, revision, alias=False): | ||||
| branches[branch] = SnapshotBranchInfo( | branches[branch] = SnapshotBranchInfo( | ||||
| name=branch, | name=branch, | ||||
| alias=alias, | alias=alias, | ||||
| revision=revision["id"], | target_type="revision", | ||||
| target=revision["id"], | |||||
| directory=revision["directory"], | directory=revision["directory"], | ||||
| date=format_utc_iso_date(revision["date"]), | date=format_utc_iso_date(revision["date"]), | ||||
| message=revision["message"], | message=revision["message"], | ||||
| url=None, | url=None, | ||||
| ) | ) | ||||
| releases_info = archive.lookup_release_multiple(release_to_branch.keys()) | releases_info = archive.lookup_release_multiple(release_to_branch.keys()) | ||||
| for release in releases_info: | for release in releases_info: | ||||
| Show All 14 Lines | for revision in revisions: | ||||
| continue | continue | ||||
| for branch in revision_to_branch[revision["id"]]: | for branch in revision_to_branch[revision["id"]]: | ||||
| _add_branch_info(branch, revision) | _add_branch_info(branch, revision) | ||||
| for release_id in revision_to_release[revision["id"]]: | for release_id in revision_to_release[revision["id"]]: | ||||
| releases[release_id]["directory"] = revision["directory"] | releases[release_id]["directory"] = revision["directory"] | ||||
| resolved_aliases = {} | resolved_aliases = {} | ||||
| for branch_alias, branch_target in branch_aliases.items(): | for branch_alias, _ in branch_aliases.items(): | ||||
| resolved_alias = archive.lookup_snapshot_alias(snapshot["id"], branch_alias) | resolved_alias = archive.lookup_snapshot_alias(snapshot["id"], branch_alias) | ||||
| resolved_aliases[branch_alias] = resolved_alias | resolved_aliases[branch_alias] = resolved_alias | ||||
| if resolved_alias is None: | if resolved_alias is None: | ||||
| continue | continue | ||||
| target_type = resolved_alias["target_type"] | target_type = resolved_alias["target_type"] | ||||
| target = resolved_alias["target"] | target = resolved_alias["target"] | ||||
| if target_type == "revision": | if target_type == "revision": | ||||
| revision = archive.lookup_revision(target) | revision = archive.lookup_revision(target) | ||||
| _add_branch_info(branch_alias, revision, alias=True) | _add_branch_info(branch_alias, revision, alias=True) | ||||
| elif target_type == "release": | elif target_type == "release": | ||||
| release = archive.lookup_release(target) | release = archive.lookup_release(target) | ||||
| _add_release_info(branch_alias, release, alias=True) | _add_release_info(branch_alias, release, alias=True) | ||||
| elif target_type in ("content", "directory"): | |||||
| branches[branch_name] = SnapshotBranchInfo( | |||||
| name=branch_alias, | |||||
| alias=True, | |||||
| target_type=target_type, | |||||
| target=target, | |||||
| date=None, | |||||
| directory=None, | |||||
| message=None, | |||||
| url=None, | |||||
| ) | |||||
| if branch_alias in branches: | if branch_alias in branches: | ||||
| branches[branch_alias]["name"] = branch_alias | branches[branch_alias]["name"] = branch_alias | ||||
| ret_branches = list(sorted(branches.values(), key=lambda b: b["name"])) | ret_branches = list(sorted(branches.values(), key=lambda b: b["name"])) | ||||
| ret_releases = list(sorted(releases.values(), key=lambda b: b["name"])) | ret_releases = list(sorted(releases.values(), key=lambda b: b["name"])) | ||||
| return ret_branches, ret_releases, resolved_aliases | return ret_branches, ret_releases, resolved_aliases | ||||
| ▲ Show 20 Lines • Show All 169 Lines • ▼ Show 20 Lines | if origin_url: | ||||
| ) | ) | ||||
| if timestamp is not None: | if timestamp is not None: | ||||
| query_params["timestamp"] = format_utc_iso_date( | query_params["timestamp"] = format_utc_iso_date( | ||||
| timestamp, "%Y-%m-%dT%H:%M:%SZ" | timestamp, "%Y-%m-%dT%H:%M:%SZ" | ||||
| ) | ) | ||||
| visit_url = reverse("browse-origin-directory", query_params=query_params) | visit_url = reverse("browse-origin-directory", query_params=query_params) | ||||
| visit_info["url"] = directory_url = visit_url | visit_info["url"] = browse_url = visit_url | ||||
| branches_url = reverse("browse-origin-branches", query_params=query_params) | branches_url = reverse("browse-origin-branches", query_params=query_params) | ||||
| releases_url = reverse("browse-origin-releases", query_params=query_params) | releases_url = reverse("browse-origin-releases", query_params=query_params) | ||||
| else: | else: | ||||
| assert snapshot_id is not None | assert snapshot_id is not None | ||||
| branches, releases, aliases = get_snapshot_content(snapshot_id) | branches, releases, aliases = get_snapshot_content(snapshot_id) | ||||
| url_args = {"snapshot_id": snapshot_id} | url_args = {"snapshot_id": snapshot_id} | ||||
| directory_url = reverse("browse-snapshot-directory", url_args=url_args) | browse_url = reverse("browse-snapshot-directory", url_args=url_args) | ||||
| branches_url = reverse("browse-snapshot-branches", url_args=url_args) | branches_url = reverse("browse-snapshot-branches", url_args=url_args) | ||||
| releases_url = reverse("browse-snapshot-releases", url_args=url_args) | releases_url = reverse("browse-snapshot-releases", url_args=url_args) | ||||
| releases = list(reversed(releases)) | releases = list(reversed(releases)) | ||||
| @django_cache() | @django_cache() | ||||
| def _get_snapshot_sizes(snapshot_id): | def _get_snapshot_sizes(snapshot_id): | ||||
| return archive.lookup_snapshot_sizes(snapshot_id) | return archive.lookup_snapshot_sizes(snapshot_id) | ||||
| snapshot_sizes = _get_snapshot_sizes(snapshot_id) | snapshot_sizes = _get_snapshot_sizes(snapshot_id) | ||||
| is_empty = (snapshot_sizes["release"] + snapshot_sizes["revision"]) == 0 | snapshot_total_size = sum( | ||||
| v for k, v in snapshot_sizes.items() if k not in ("alias", "branch") | |||||
| ) | |||||
| is_empty = snapshot_total_size == 0 | |||||
| swh_snp_id = str( | swh_snp_id = str( | ||||
| CoreSWHID(object_type=ObjectType.SNAPSHOT, object_id=hash_to_bytes(snapshot_id)) | CoreSWHID(object_type=ObjectType.SNAPSHOT, object_id=hash_to_bytes(snapshot_id)) | ||||
| ) | ) | ||||
| if visit_info: | if visit_info: | ||||
| timestamp = format_utc_iso_date(visit_info["date"]) | timestamp = format_utc_iso_date(visit_info["date"]) | ||||
| if origin_info: | if origin_info: | ||||
| browse_view_name = f"browse-origin-{browse_context}" | browse_view_name = f"browse-origin-{browse_context}" | ||||
| else: | else: | ||||
| browse_view_name = f"browse-snapshot-{browse_context}" | browse_view_name = f"browse-snapshot-{browse_context}" | ||||
| release_id = None | release_id = None | ||||
| root_directory = None | root_directory = None | ||||
| snapshot_total_size = snapshot_sizes["release"] + snapshot_sizes["revision"] | |||||
| if path is not None: | if path is not None: | ||||
| query_params["path"] = path | query_params["path"] = path | ||||
| if snapshot_total_size and revision_id is not None: | if snapshot_total_size and revision_id is not None: | ||||
| # browse specific revision for a snapshot requested | # browse specific revision for a snapshot requested | ||||
| revision = archive.lookup_revision(revision_id) | revision = archive.lookup_revision(revision_id) | ||||
| root_directory = revision["directory"] | root_directory = revision["directory"] | ||||
| branches.append( | branches.append( | ||||
| SnapshotBranchInfo( | SnapshotBranchInfo( | ||||
| name=revision_id, | name=revision_id, | ||||
| alias=False, | alias=False, | ||||
| revision=revision_id, | target_type="revision", | ||||
| target=revision_id, | |||||
| directory=root_directory, | directory=root_directory, | ||||
| date=revision["date"], | date=revision["date"], | ||||
| message=revision["message"], | message=revision["message"], | ||||
| url=None, | url=None, | ||||
| ) | ) | ||||
| ) | ) | ||||
| query_params["revision"] = revision_id | query_params["revision"] = revision_id | ||||
| elif snapshot_total_size and release_name: | elif snapshot_total_size and release_name: | ||||
| Show All 31 Lines | elif snapshot_total_size: | ||||
| snapshot_id, | snapshot_id, | ||||
| snapshot_sizes, | snapshot_sizes, | ||||
| origin_info, | origin_info, | ||||
| timestamp, | timestamp, | ||||
| visit_id, | visit_id, | ||||
| ) | ) | ||||
| else: | else: | ||||
| branch_name = branch["name"] | branch_name = branch["name"] | ||||
| revision_id = branch["revision"] | if branch["target_type"] == "revision": | ||||
| revision_id = branch["target"] | |||||
| root_directory = branch["directory"] | root_directory = branch["directory"] | ||||
| elif branch["target_type"] == "directory": | |||||
| root_directory = branch["target"] | |||||
| elif head is not None: | elif head is not None: | ||||
| # otherwise, browse branch targeted by the HEAD alias if it exists | # otherwise, browse branch targeted by the HEAD alias if it exists | ||||
| if head["target_type"] in ("content", "directory", "revision"): | |||||
| branch_name = "HEAD" | |||||
| if head["target_type"] == "revision": | if head["target_type"] == "revision": | ||||
| # HEAD alias targets a revision | # HEAD alias targets a revision | ||||
| head_rev = archive.lookup_revision(head["target"]) | head_rev = archive.lookup_revision(head["target"]) | ||||
| branch_name = "HEAD" | |||||
| revision_id = head_rev["id"] | revision_id = head_rev["id"] | ||||
| root_directory = head_rev["directory"] | root_directory = head_rev["directory"] | ||||
| else: | elif head["target_type"] == "directory": | ||||
| # HEAD alias targets a directory | |||||
| root_directory = head["target"] | |||||
| elif head["target_type"] == "release": | |||||
| # HEAD alias targets a release | # HEAD alias targets a release | ||||
| release_name = archive.lookup_release(head["target"])["name"] | release_name = archive.lookup_release(head["target"])["name"] | ||||
| head_rel = _get_release(releases, release_name, snapshot_id) | head_rel = _get_release(releases, release_name, snapshot_id) | ||||
| if head_rel is None: | if head_rel is None: | ||||
| _branch_not_found( | _branch_not_found( | ||||
| "release", | "release", | ||||
| str(release_name), | str(release_name), | ||||
| snapshot_id, | snapshot_id, | ||||
| Show All 9 Lines | elif snapshot_total_size: | ||||
| elif head_rel["target_type"] == "directory": | elif head_rel["target_type"] == "directory": | ||||
| root_directory = head_rel["target"] | root_directory = head_rel["target"] | ||||
| if head_rel is not None: | if head_rel is not None: | ||||
| release_id = head_rel["id"] | release_id = head_rel["id"] | ||||
| elif branches: | elif branches: | ||||
| # fallback to browse first branch otherwise | # fallback to browse first branch otherwise | ||||
| branch = branches[0] | branch = branches[0] | ||||
| branch_name = branch["name"] | branch_name = branch["name"] | ||||
| revision_id = branch["revision"] | revision_id = ( | ||||
| branch["target"] if branch["target_type"] == "revision" else None | |||||
| ) | |||||
| if branch["target_type"] == "revision": | |||||
| root_directory = branch["directory"] | root_directory = branch["directory"] | ||||
| elif branch["target_type"] == "directory": | |||||
| root_directory = branch["target"] | |||||
| elif releases: | elif releases: | ||||
| # fallback to browse last release otherwise | # fallback to browse last release otherwise | ||||
| release = releases[-1] | release = releases[-1] | ||||
| if release["target_type"] == "revision": | if release["target_type"] == "revision": | ||||
| revision = archive.lookup_revision(release["target"]) | revision = archive.lookup_revision(release["target"]) | ||||
| root_directory = revision["directory"] | root_directory = revision["directory"] | ||||
| revision_id = release["target"] | revision_id = release["target"] | ||||
| elif release["target_type"] == "directory": | elif release["target_type"] == "directory": | ||||
| root_directory = release["target"] | root_directory = release["target"] | ||||
| release_id = release["id"] | release_id = release["id"] | ||||
| release_name = release["name"] | release_name = release["name"] | ||||
| for b in branches: | for b in branches: | ||||
| branch_query_params = dict(query_params) | branch_query_params = dict(query_params) | ||||
| branch_query_params.pop("release", None) | branch_query_params.pop("release", None) | ||||
| if b["name"] != b["revision"]: | if b["name"] != b["target"]: | ||||
| branch_query_params.pop("revision", None) | branch_query_params.pop("revision", None) | ||||
| branch_query_params["branch"] = b["name"] | branch_query_params["branch"] = b["name"] | ||||
| b["url"] = reverse( | b["url"] = reverse( | ||||
| browse_view_name, url_args=url_args, query_params=branch_query_params | browse_view_name, url_args=url_args, query_params=branch_query_params | ||||
| ) | ) | ||||
| for r in releases: | for r in releases: | ||||
| release_query_params = dict(query_params) | release_query_params = dict(query_params) | ||||
| Show All 19 Lines | if revision_id: | ||||
| ) | ) | ||||
| if revision_info["message"]: | if revision_info["message"]: | ||||
| message_lines = revision_info["message"].split("\n") | message_lines = revision_info["message"].split("\n") | ||||
| revision_info["message_header"] = message_lines[0] | revision_info["message_header"] = message_lines[0] | ||||
| else: | else: | ||||
| revision_info["message_header"] = "" | revision_info["message_header"] = "" | ||||
| snapshot_context = SnapshotContext( | snapshot_context = SnapshotContext( | ||||
| directory_url=directory_url, | browse_url=browse_url, | ||||
| branch=branch_name, | branch=branch_name, | ||||
| branch_alias=branch_name in aliases, | branch_alias=branch_name in aliases, | ||||
| branches=branches, | branches=branches, | ||||
| branches_url=branches_url, | branches_url=branches_url, | ||||
| is_empty=is_empty, | is_empty=is_empty, | ||||
| origin_info=origin_info, | origin_info=origin_info, | ||||
| origin_visits_url=origin_visits_url, | origin_visits_url=origin_visits_url, | ||||
| release=release_name, | release=release_name, | ||||
| ▲ Show 20 Lines • Show All 85 Lines • ▼ Show 20 Lines | snapshot_context = get_snapshot_context( | ||||
| path=path, | path=path, | ||||
| browse_context="directory", | browse_context="directory", | ||||
| branch_name=request.GET.get("branch"), | branch_name=request.GET.get("branch"), | ||||
| release_name=request.GET.get("release"), | release_name=request.GET.get("release"), | ||||
| revision_id=request.GET.get("revision"), | revision_id=request.GET.get("revision"), | ||||
| ) | ) | ||||
| root_directory = snapshot_context["root_directory"] | root_directory = snapshot_context["root_directory"] | ||||
| if root_directory is None and snapshot_context["branch"] is not None: | |||||
| branch_info = [ | |||||
| branch | |||||
| for branch in snapshot_context["branches"] | |||||
| if branch["name"] == snapshot_context["branch"] | |||||
| ] | |||||
| # special case where the branch to browse targets a content instead of a directory | |||||
| if branch_info and branch_info[0]["target_type"] == "content": | |||||
| # redirect to browse content view | |||||
| if "origin_url" not in snapshot_context["query_params"]: | |||||
| snapshot_id = snapshot_context["snapshot_id"] | |||||
| snapshot_context["query_params"]["snapshot"] = snapshot_id | |||||
| return redirect( | |||||
| reverse( | |||||
| "browse-content", | |||||
| url_args={"query_string": f"sha1_git:{branch_info[0]['target']}"}, | |||||
| query_params=snapshot_context["query_params"], | |||||
| ) | |||||
| ) | |||||
| sha1_git = root_directory | sha1_git = root_directory | ||||
| error_info: Dict[str, Any] = { | error_info: Dict[str, Any] = { | ||||
| "status_code": 200, | "status_code": 200, | ||||
| "description": None, | "description": None, | ||||
| } | } | ||||
| if root_directory and path: | if root_directory and path: | ||||
| try: | try: | ||||
| dir_info = archive.lookup_directory_with_path(root_directory, path) | dir_info = archive.lookup_directory_with_path(root_directory, path) | ||||
| ▲ Show 20 Lines • Show All 392 Lines • ▼ Show 20 Lines | ) -> HttpResponse: | ||||
| for branch in displayed_branches: | for branch in displayed_branches: | ||||
| rev_query_params = {} | rev_query_params = {} | ||||
| if origin_info: | if origin_info: | ||||
| rev_query_params["origin_url"] = origin_info["url"] | rev_query_params["origin_url"] = origin_info["url"] | ||||
| revision_url = reverse( | revision_url = reverse( | ||||
| "browse-revision", | "browse-revision", | ||||
| url_args={"sha1_git": branch["revision"]}, | url_args={"sha1_git": branch["target"]}, | ||||
| query_params=query_params, | query_params=query_params, | ||||
| ) | ) | ||||
| query_params["branch"] = branch["name"] | query_params["branch"] = branch["name"] | ||||
| directory_url = reverse( | directory_url = reverse( | ||||
| browse_view_name, url_args=url_args, query_params=query_params | browse_view_name, url_args=url_args, query_params=query_params | ||||
| ) | ) | ||||
| del query_params["branch"] | del query_params["branch"] | ||||
| ▲ Show 20 Lines • Show All 201 Lines • Show Last 20 Lines | |||||