Changeset View
Changeset View
Standalone View
Standalone View
swh/web/browse/snapshot_context.py
| # Copyright (C) 2018-2019 The Software Heritage developers | # Copyright (C) 2018-2019 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 copy import copy | |||||
| from typing import Any, Dict, List, Optional, Union, Tuple | |||||
| from django.core.cache import cache | from django.core.cache import cache | ||||
| from django.shortcuts import render | from django.shortcuts import render | ||||
| from django.template.defaultfilters import filesizeformat | from django.template.defaultfilters import filesizeformat | ||||
| from django.utils.html import escape | from django.utils.html import escape | ||||
| import sentry_sdk | import sentry_sdk | ||||
| Show All 13 Lines | from swh.web.browse.utils import ( | ||||
| get_readme_to_display, | get_readme_to_display, | ||||
| get_swh_persistent_ids, | get_swh_persistent_ids, | ||||
| gen_snapshot_link, | gen_snapshot_link, | ||||
| ) | ) | ||||
| from swh.web.common import service, highlightjs | from swh.web.common import service, highlightjs | ||||
| from swh.web.common.exc import handle_view_exception, NotFoundExc | from swh.web.common.exc import handle_view_exception, NotFoundExc | ||||
| from swh.web.common.origin_visits import get_origin_visit | from swh.web.common.origin_visits import get_origin_visit | ||||
| from swh.web.common.typing import ( | |||||
| OriginInfo, | |||||
| QueryParameters, | |||||
| SnapshotBranchInfo, | |||||
| SnapshotReleaseInfo, | |||||
| SnapshotContext, | |||||
| ) | |||||
| from swh.web.common.utils import ( | from swh.web.common.utils import ( | ||||
| reverse, | reverse, | ||||
| gen_path_info, | gen_path_info, | ||||
| format_utc_iso_date, | format_utc_iso_date, | ||||
| swh_object_icons, | swh_object_icons, | ||||
| ) | ) | ||||
| from swh.web.config import get_config | from swh.web.config import get_config | ||||
| ▲ Show 20 Lines • Show All 103 Lines • ▼ Show 20 Lines | else: | ||||
| msg = ( | msg = ( | ||||
| "%s %s associated to visit with" | "%s %s associated to visit with" | ||||
| " timestamp %s for origin with " | " timestamp %s for origin with " | ||||
| "url %s not found!" % (branch_type, branch, timestamp, origin_info["url"]) | "url %s not found!" % (branch_type, branch, timestamp, origin_info["url"]) | ||||
| ) | ) | ||||
| raise NotFoundExc(escape(msg)) | raise NotFoundExc(escape(msg)) | ||||
| def process_snapshot_branches(snapshot): | def process_snapshot_branches( | ||||
| snapshot: Dict[str, Any] | |||||
| ) -> Tuple[List[SnapshotBranchInfo], List[SnapshotReleaseInfo]]: | |||||
| """ | """ | ||||
| Process a dictionary describing snapshot branches: extract those | Process a dictionary describing snapshot branches: extract those | ||||
| targeting revisions and releases, put them in two different lists, | targeting revisions and releases, put them in two different lists, | ||||
| then sort those lists in lexicographical order of the branches' names. | then sort those lists in lexicographical order of the branches' names. | ||||
| Args: | Args: | ||||
| snapshot_branches (dict): A dict describing the branches of a snapshot | snapshot: A dict describing a snapshot as returned for instance by | ||||
| as returned for instance by | |||||
| :func:`swh.web.common.service.lookup_snapshot` | :func:`swh.web.common.service.lookup_snapshot` | ||||
| Returns: | Returns: | ||||
| tuple: A tuple whose first member is the sorted list of branches | A tuple whose first member is the sorted list of branches | ||||
| targeting revisions and second member the sorted list of branches | targeting revisions and second member the sorted list of branches | ||||
| targeting releases | targeting releases | ||||
| """ | """ | ||||
| snapshot_branches = snapshot["branches"] | snapshot_branches = snapshot["branches"] | ||||
| branches = {} | branches: Dict[str, SnapshotBranchInfo] = {} | ||||
| branch_aliases = {} | branch_aliases: Dict[str, str] = {} | ||||
| releases = {} | releases: Dict[str, SnapshotReleaseInfo] = {} | ||||
| revision_to_branch = defaultdict(set) | revision_to_branch = defaultdict(set) | ||||
| 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 == "revision": | ||||
| branches[branch_name] = { | branches[branch_name] = SnapshotBranchInfo( | ||||
| "name": branch_name, | name=branch_name, | ||||
| "revision": target_id, | revision=target_id, | ||||
| } | date=None, | ||||
| directory=None, | |||||
| message=None, | |||||
| url=None, | |||||
| ) | |||||
| 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 _enrich_release_branch(branch, release): | def _add_release_info(branch, release): | ||||
| releases[branch] = { | releases[branch] = SnapshotReleaseInfo( | ||||
| "name": release["name"], | name=release["name"], | ||||
| "branch_name": branch, | branch_name=branch, | ||||
| "date": format_utc_iso_date(release["date"]), | date=format_utc_iso_date(release["date"]), | ||||
| "id": release["id"], | directory=None, | ||||
| "message": release["message"], | id=release["id"], | ||||
| "target_type": release["target_type"], | message=release["message"], | ||||
| "target": release["target"], | target_type=release["target_type"], | ||||
| } | target=release["target"], | ||||
| url=None, | |||||
| def _enrich_revision_branch(branch, revision): | ) | ||||
| branches[branch].update( | |||||
| { | def _add_branch_info(branch, revision): | ||||
| "revision": revision["id"], | branches[branch] = SnapshotBranchInfo( | ||||
| "directory": revision["directory"], | name=branch, | ||||
| "date": format_utc_iso_date(revision["date"]), | revision=revision["id"], | ||||
| "message": revision["message"], | directory=revision["directory"], | ||||
| } | date=format_utc_iso_date(revision["date"]), | ||||
| message=revision["message"], | |||||
| url=None, | |||||
| ) | ) | ||||
| releases_info = service.lookup_release_multiple(release_to_branch.keys()) | releases_info = service.lookup_release_multiple(release_to_branch.keys()) | ||||
| for release in releases_info: | for release in releases_info: | ||||
| branches_to_update = release_to_branch[release["id"]] | branches_to_update = release_to_branch[release["id"]] | ||||
| for branch in branches_to_update: | for branch in branches_to_update: | ||||
| _enrich_release_branch(branch, release) | _add_release_info(branch, release) | ||||
| if release["target_type"] == "revision": | if release["target_type"] == "revision": | ||||
| revision_to_release[release["target"]].update(branches_to_update) | revision_to_release[release["target"]].update(branches_to_update) | ||||
| revisions = service.lookup_revision_multiple( | revisions = service.lookup_revision_multiple( | ||||
| set(revision_to_branch.keys()) | set(revision_to_release.keys()) | set(revision_to_branch.keys()) | set(revision_to_release.keys()) | ||||
| ) | ) | ||||
| for revision in revisions: | for revision in revisions: | ||||
| if not revision: | if not revision: | ||||
| continue | continue | ||||
| for branch in revision_to_branch[revision["id"]]: | for branch in revision_to_branch[revision["id"]]: | ||||
| _enrich_revision_branch(branch, revision) | _add_branch_info(branch, revision) | ||||
| for release in revision_to_release[revision["id"]]: | for release in revision_to_release[revision["id"]]: | ||||
| releases[release]["directory"] = revision["directory"] | releases[release]["directory"] = revision["directory"] | ||||
| for branch_alias, branch_target in branch_aliases.items(): | for branch_alias, branch_target in branch_aliases.items(): | ||||
| if branch_target in branches: | if branch_target in branches: | ||||
| branches[branch_alias] = dict(branches[branch_target]) | branches[branch_alias] = copy(branches[branch_target]) | ||||
| else: | else: | ||||
| snp = service.lookup_snapshot( | snp = service.lookup_snapshot( | ||||
| snapshot["id"], branches_from=branch_target, branches_count=1 | snapshot["id"], branches_from=branch_target, branches_count=1 | ||||
| ) | ) | ||||
| if snp and branch_target in snp["branches"]: | if snp and branch_target in snp["branches"]: | ||||
| if snp["branches"][branch_target] is None: | if snp["branches"][branch_target] is None: | ||||
| continue | continue | ||||
| target_type = snp["branches"][branch_target]["target_type"] | target_type = snp["branches"][branch_target]["target_type"] | ||||
| target = snp["branches"][branch_target]["target"] | target = snp["branches"][branch_target]["target"] | ||||
| if target_type == "revision": | if target_type == "revision": | ||||
| branches[branch_alias] = snp["branches"][branch_target] | branches[branch_alias] = snp["branches"][branch_target] | ||||
| revision = service.lookup_revision(target) | revision = service.lookup_revision(target) | ||||
| _enrich_revision_branch(branch_alias, revision) | _add_branch_info(branch_alias, revision) | ||||
| elif target_type == "release": | elif target_type == "release": | ||||
| release = service.lookup_release(target) | release = service.lookup_release(target) | ||||
| _enrich_release_branch(branch_alias, release) | _add_release_info(branch_alias, release) | ||||
| 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 | return ret_branches, ret_releases | ||||
| def get_snapshot_content(snapshot_id): | def get_snapshot_content( | ||||
| snapshot_id: str, | |||||
| ) -> Tuple[List[SnapshotBranchInfo], List[SnapshotReleaseInfo]]: | |||||
| """Returns the lists of branches and releases | """Returns the lists of branches and releases | ||||
| associated to a swh snapshot. | associated to a swh snapshot. | ||||
| That list is put in cache in order to speedup the navigation | That list is put in cache in order to speedup the navigation | ||||
| in the swh-web/browse ui. | in the swh-web/browse ui. | ||||
| .. warning:: At most 1000 branches contained in the snapshot | .. warning:: At most 1000 branches contained in the snapshot | ||||
| will be returned for performance reasons. | will be returned for performance reasons. | ||||
| Args: | Args: | ||||
| snapshot_id (str): hexadecimal representation of the snapshot | snapshot_id: hexadecimal representation of the snapshot identifier | ||||
| identifier | |||||
| Returns: | Returns: | ||||
| A tuple with two members. The first one is a list of dict describing | A tuple with two members. The first one is a list of dict describing | ||||
| the snapshot branches. The second one is a list of dict describing the | the snapshot branches. The second one is a list of dict describing the | ||||
| snapshot releases. | snapshot releases. | ||||
| Raises: | Raises: | ||||
| NotFoundExc if the snapshot does not exist | NotFoundExc if the snapshot does not exist | ||||
| """ | """ | ||||
| cache_entry_id = "swh_snapshot_%s" % snapshot_id | cache_entry_id = "swh_snapshot_%s" % snapshot_id | ||||
| cache_entry = cache.get(cache_entry_id) | cache_entry = cache.get(cache_entry_id) | ||||
| if cache_entry: | if cache_entry: | ||||
| return cache_entry["branches"], cache_entry["releases"] | return cache_entry["branches"], cache_entry["releases"] | ||||
| branches = [] | branches: List[SnapshotBranchInfo] = [] | ||||
| releases = [] | releases: List[SnapshotReleaseInfo] = [] | ||||
| snapshot_content_max_size = get_config()["snapshot_content_max_size"] | snapshot_content_max_size = get_config()["snapshot_content_max_size"] | ||||
| if snapshot_id: | if snapshot_id: | ||||
| snapshot = service.lookup_snapshot( | snapshot = service.lookup_snapshot( | ||||
| snapshot_id, branches_count=snapshot_content_max_size | snapshot_id, branches_count=snapshot_content_max_size | ||||
| ) | ) | ||||
| branches, releases = process_snapshot_branches(snapshot) | branches, releases = process_snapshot_branches(snapshot) | ||||
| cache.set(cache_entry_id, {"branches": branches, "releases": releases,}) | cache.set(cache_entry_id, {"branches": branches, "releases": releases,}) | ||||
| return branches, releases | return branches, releases | ||||
| def get_origin_visit_snapshot( | def get_origin_visit_snapshot( | ||||
| origin_info, visit_ts=None, visit_id=None, snapshot_id=None | origin_info: OriginInfo, | ||||
| ): | visit_ts: Optional[Union[int, str]] = None, | ||||
| """Returns the lists of branches and releases | visit_id: Optional[int] = None, | ||||
| associated to a swh origin for a given visit. | snapshot_id: Optional[str] = None, | ||||
| The visit is expressed by a timestamp. In the latter case, | ) -> Tuple[List[SnapshotBranchInfo], List[SnapshotReleaseInfo]]: | ||||
| the closest visit from the provided timestamp will be used. | """Returns the lists of branches and releases associated to an origin for | ||||
| a given visit. | |||||
| The visit is expressed by either: | |||||
| * a snapshot identifier | |||||
| * a timestamp, if no visit with that exact timestamp is found, | |||||
| the closest one from the provided timestamp will be used. | |||||
| If no visit parameter is provided, it returns the list of branches | If no visit parameter is provided, it returns the list of branches | ||||
| found for the latest visit. | found for the latest visit. | ||||
| That list is put in cache in order to speedup the navigation | That list is put in cache in order to speedup the navigation | ||||
| in the swh-web/browse ui. | in the swh-web/browse ui. | ||||
| .. warning:: At most 1000 branches contained in the snapshot | .. warning:: At most 1000 branches contained in the snapshot | ||||
| will be returned for performance reasons. | will be returned for performance reasons. | ||||
| Args: | Args: | ||||
| origin_info (dict): a dict filled with origin information | origin_info: a dict filled with origin information | ||||
| (id, url, type) | visit_ts: an ISO date string or Unix timestamp to parse | ||||
| visit_ts (int or str): an ISO date string or Unix timestamp to parse | visit_id: visit id for disambiguation in case several visits have | ||||
| visit_id (int): optional visit id for disambiguation in case | the same timestamp | ||||
| several visits have the same timestamp | snapshot_id: if provided, visit associated to the snapshot will be processed | ||||
| Returns: | Returns: | ||||
| A tuple with two members. The first one is a list of dict describing | A tuple with two members. The first one is a list of dict describing | ||||
| the origin branches for the given visit. | the origin branches for the given visit. | ||||
| The second one is a list of dict describing the origin releases | The second one is a list of dict describing the origin releases | ||||
| for the given visit. | for the given visit. | ||||
| Raises: | Raises: | ||||
| NotFoundExc if the origin or its visit are not found | NotFoundExc if the origin or its visit are not found | ||||
| """ | """ | ||||
| visit_info = get_origin_visit(origin_info, visit_ts, visit_id, snapshot_id) | visit_info = get_origin_visit(origin_info, visit_ts, visit_id, snapshot_id) | ||||
| return get_snapshot_content(visit_info["snapshot"]) | return get_snapshot_content(visit_info["snapshot"]) | ||||
| def get_snapshot_context( | def get_snapshot_context( | ||||
| snapshot_id=None, origin_url=None, timestamp=None, visit_id=None | snapshot_id: Optional[str] = None, | ||||
| ): | origin_url: Optional[str] = None, | ||||
| timestamp: Optional[str] = None, | |||||
| visit_id: Optional[int] = None, | |||||
| branch_name: Optional[str] = None, | |||||
| release_name: Optional[str] = None, | |||||
| revision_id: Optional[str] = None, | |||||
| path: Optional[str] = None, | |||||
| browse_context: str = "directory", | |||||
| ) -> SnapshotContext: | |||||
| """ | """ | ||||
| Utility function to compute relevant information when navigating | Utility function to compute relevant information when navigating | ||||
| the archive in a snapshot context. The snapshot is either | the archive in a snapshot context. The snapshot is either | ||||
| referenced by its id or it will be retrieved from an origin visit. | referenced by its id or it will be retrieved from an origin visit. | ||||
| Args: | Args: | ||||
| snapshot_id (str): hexadecimal representation of a snapshot identifier, | snapshot_id: hexadecimal representation of a snapshot identifier | ||||
| all other parameters will be ignored if it is provided | origin_url: an origin_url | ||||
| origin_url (str): the origin_url | timestamp: a datetime string for retrieving the closest | ||||
| (e.g. https://github.com/(user)/(repo)/) | |||||
| timestamp (str): a datetime string for retrieving the closest | |||||
| visit of the origin | visit of the origin | ||||
| visit_id (int): optional visit id for disambiguation in case | visit_id: optional visit id for disambiguation in case | ||||
| of several visits with the same timestamp | of several visits with the same timestamp | ||||
| branch_name: optional branch name set when browsing the snapshot in | |||||
| that scope (will default to "HEAD" if not provided) | |||||
| release_name: optional release name set when browsing the snapshot in | |||||
| that scope | |||||
| revision_id: optional revision identifier set when browsing the snapshot in | |||||
| that scope | |||||
| path: optional path of the object currently browsed in the snapshot | |||||
| browse_context: indicates which type of object is currently browsed | |||||
| Returns: | Returns: | ||||
| A dict with the following entries: | A dict filled with snapshot context information. | ||||
| * origin_info: dict containing origin information | |||||
| * visit_info: dict containing visit information | |||||
| * branches: the list of branches for the origin found | |||||
| during the visit | |||||
| * releases: the list of releases for the origin found | |||||
| during the visit | |||||
| * origin_browse_url: the url to browse the origin | |||||
| * origin_branches_url: the url to browse the origin branches | |||||
| * origin_releases_url': the url to browse the origin releases | |||||
| * origin_visit_url: the url to browse the snapshot of the origin | |||||
| found during the visit | |||||
| * url_args: dict containing url arguments to use when browsing in | |||||
| the context of the origin and its visit | |||||
| Raises: | Raises: | ||||
| swh.web.common.exc.NotFoundExc: if no snapshot is found for the visit | swh.web.common.exc.NotFoundExc: if no snapshot is found for the visit | ||||
| of an origin. | of an origin. | ||||
| """ | """ | ||||
| assert origin_url is not None or snapshot_id is not None | |||||
| origin_info = None | origin_info = None | ||||
| visit_info = None | visit_info = None | ||||
| url_args = None | url_args = {} | ||||
| query_params = {} | query_params: QueryParameters = {} | ||||
| branches = [] | origin_visits_url = None | ||||
| releases = [] | |||||
| browse_url = None | |||||
| visit_url = None | |||||
| branches_url = None | |||||
| releases_url = None | |||||
| swh_type = "snapshot" | |||||
| if origin_url: | if origin_url: | ||||
| swh_type = "origin" | |||||
| origin_info = service.lookup_origin({"url": origin_url}) | origin_info = service.lookup_origin({"url": origin_url}) | ||||
| visit_info = get_origin_visit(origin_info, timestamp, visit_id, snapshot_id) | visit_info = get_origin_visit(origin_info, timestamp, visit_id, snapshot_id) | ||||
| fmt_date = format_utc_iso_date(visit_info["date"]) | formatted_date = format_utc_iso_date(visit_info["date"]) | ||||
| visit_info["fmt_date"] = fmt_date | visit_info["formatted_date"] = formatted_date | ||||
| snapshot_id = visit_info["snapshot"] | snapshot_id = visit_info["snapshot"] | ||||
| if not snapshot_id: | if not snapshot_id: | ||||
| raise NotFoundExc( | raise NotFoundExc( | ||||
| "No snapshot associated to the visit of origin " | "No snapshot associated to the visit of origin " | ||||
| "%s on %s" % (escape(origin_url), fmt_date) | "%s on %s" % (escape(origin_url), formatted_date) | ||||
| ) | ) | ||||
| # provided timestamp is not necessarily equals to the one | # provided timestamp is not necessarily equals to the one | ||||
| # of the retrieved visit, so get the exact one in order | # of the retrieved visit, so get the exact one in order | ||||
| # use it in the urls generated below | # to use it in the urls generated below | ||||
| if timestamp: | if timestamp: | ||||
| timestamp = visit_info["date"] | timestamp = visit_info["date"] | ||||
| branches, releases = get_origin_visit_snapshot( | branches, releases = get_origin_visit_snapshot( | ||||
| origin_info, timestamp, visit_id, snapshot_id | origin_info, timestamp, visit_id, snapshot_id | ||||
| ) | ) | ||||
| url_args = {"origin_url": origin_info["url"]} | url_args = {"origin_url": origin_info["url"]} | ||||
| query_params = {"visit_id": visit_id} | query_params = {} | ||||
| if visit_id is not None: | |||||
| query_params["visit_id"] = visit_id | |||||
| browse_url = reverse("browse-origin-visits", url_args=url_args) | origin_visits_url = reverse("browse-origin-visits", url_args=url_args) | ||||
| if timestamp: | if timestamp: | ||||
| url_args["timestamp"] = format_utc_iso_date(timestamp, "%Y-%m-%dT%H:%M:%S") | url_args["timestamp"] = format_utc_iso_date(timestamp, "%Y-%m-%dT%H:%M:%SZ") | ||||
| visit_url = reverse( | visit_url = reverse( | ||||
| "browse-origin-directory", url_args=url_args, query_params=query_params | "browse-origin-directory", url_args=url_args, query_params=query_params | ||||
| ) | ) | ||||
| visit_info["url"] = visit_url | visit_info["url"] = visit_url | ||||
| branches_url = reverse( | branches_url = reverse( | ||||
| "browse-origin-branches", url_args=url_args, query_params=query_params | "browse-origin-branches", url_args=url_args, query_params=query_params | ||||
| ) | ) | ||||
| releases_url = reverse( | releases_url = reverse( | ||||
| "browse-origin-releases", url_args=url_args, query_params=query_params | "browse-origin-releases", url_args=url_args, query_params=query_params | ||||
| ) | ) | ||||
| elif snapshot_id: | else: | ||||
| assert snapshot_id is not None | |||||
| branches, releases = get_snapshot_content(snapshot_id) | branches, releases = get_snapshot_content(snapshot_id) | ||||
| url_args = {"snapshot_id": snapshot_id} | url_args = {"snapshot_id": snapshot_id} | ||||
| browse_url = reverse("browse-snapshot", 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)) | ||||
| snapshot_sizes = service.lookup_snapshot_sizes(snapshot_id) | snapshot_sizes = service.lookup_snapshot_sizes(snapshot_id) | ||||
| is_empty = sum(snapshot_sizes.values()) == 0 | is_empty = sum(snapshot_sizes.values()) == 0 | ||||
| swh_snp_id = persistent_identifier("snapshot", snapshot_id) | swh_snp_id = persistent_identifier("snapshot", snapshot_id) | ||||
| return { | if visit_info: | ||||
| "swh_type": swh_type, | timestamp = format_utc_iso_date(visit_info["date"]) | ||||
| "swh_object_id": swh_snp_id, | |||||
| "snapshot_id": snapshot_id, | |||||
| "snapshot_sizes": snapshot_sizes, | |||||
| "is_empty": is_empty, | |||||
| "origin_info": origin_info, | |||||
| "visit_info": visit_info, | |||||
| "branches": branches, | |||||
| "releases": releases, | |||||
| "branch": None, | |||||
| "release": None, | |||||
| "browse_url": browse_url, | |||||
| "branches_url": branches_url, | |||||
| "releases_url": releases_url, | |||||
| "url_args": url_args, | |||||
| "query_params": query_params, | |||||
| } | |||||
| def _process_snapshot_request( | |||||
| request, | |||||
| snapshot_id=None, | |||||
| origin_url=None, | |||||
| timestamp=None, | |||||
| path=None, | |||||
| browse_context="directory", | |||||
| ): | |||||
| """ | |||||
| Utility function to perform common input request processing | |||||
| for snapshot context views. | |||||
| """ | |||||
| visit_id = request.GET.get("visit_id", None) | |||||
| snapshot_context = get_snapshot_context( | |||||
| snapshot_id, origin_url, timestamp, visit_id | |||||
| ) | |||||
| swh_type = snapshot_context["swh_type"] | |||||
| origin_info = snapshot_context["origin_info"] | |||||
| branches = snapshot_context["branches"] | |||||
| releases = snapshot_context["releases"] | |||||
| url_args = snapshot_context["url_args"] | |||||
| query_params = snapshot_context["query_params"] | |||||
| if snapshot_context["visit_info"]: | |||||
| timestamp = format_utc_iso_date( | |||||
| snapshot_context["visit_info"]["date"], "%Y-%m-%dT%H:%M:%SZ" | |||||
| ) | |||||
| snapshot_context["timestamp"] = format_utc_iso_date( | |||||
| snapshot_context["visit_info"]["date"] | |||||
| ) | |||||
| browse_view_name = "browse-" + swh_type + "-" + browse_context | if origin_info: | ||||
| browse_view_name = f"browse-origin-{browse_context}" | |||||
| else: | |||||
| browse_view_name = f"browse-snapshot-{browse_context}" | |||||
| root_sha1_git = None | |||||
| revision_id = request.GET.get("revision", None) | |||||
| release_name = request.GET.get("release", None) | |||||
| release_id = None | release_id = None | ||||
| branch_name = None | root_directory = None | ||||
| snapshot_sizes = snapshot_context["snapshot_sizes"] | |||||
| snapshot_total_size = sum(snapshot_sizes.values()) | snapshot_total_size = sum(snapshot_sizes.values()) | ||||
| if snapshot_total_size and revision_id: | if snapshot_total_size and revision_id is not None: | ||||
| revision = service.lookup_revision(revision_id) | revision = service.lookup_revision(revision_id) | ||||
| root_sha1_git = revision["directory"] | root_directory = revision["directory"] | ||||
| branches.append( | branches.append( | ||||
| { | SnapshotBranchInfo( | ||||
| "name": revision_id, | name=revision_id, | ||||
| "revision": revision_id, | revision=revision_id, | ||||
| "directory": root_sha1_git, | directory=root_directory, | ||||
| "url": None, | date=revision["date"], | ||||
| } | message=revision["message"], | ||||
| url=None, | |||||
| ) | |||||
| ) | ) | ||||
| branch_name = revision_id | branch_name = revision_id | ||||
| query_params["revision"] = revision_id | query_params["revision"] = revision_id | ||||
| elif snapshot_total_size and release_name: | elif snapshot_total_size and release_name: | ||||
| release = _get_release(releases, release_name, snapshot_context["snapshot_id"]) | release = _get_release(releases, release_name, snapshot_id) | ||||
| try: | try: | ||||
| root_sha1_git = release["directory"] | root_directory = release["directory"] | ||||
| revision_id = release["target"] | revision_id = release["target"] | ||||
| release_id = release["id"] | release_id = release["id"] | ||||
| query_params["release"] = release_name | query_params["release"] = release_name | ||||
| except Exception as exc: | except Exception as exc: | ||||
| sentry_sdk.capture_exception(exc) | sentry_sdk.capture_exception(exc) | ||||
| _branch_not_found( | _branch_not_found( | ||||
| "release", | "release", | ||||
| release_name, | release_name, | ||||
| snapshot_id, | snapshot_id, | ||||
| snapshot_sizes, | snapshot_sizes, | ||||
| origin_info, | origin_info, | ||||
| timestamp, | timestamp, | ||||
| visit_id, | visit_id, | ||||
| ) | ) | ||||
| elif snapshot_total_size: | elif snapshot_total_size: | ||||
| branch_name = request.GET.get("branch", None) | |||||
| if branch_name: | if branch_name: | ||||
| query_params["branch"] = branch_name | query_params["branch"] = branch_name | ||||
| branch = _get_branch( | branch = _get_branch(branches, branch_name or "HEAD", snapshot_id) | ||||
| branches, branch_name or "HEAD", snapshot_context["snapshot_id"] | |||||
| ) | |||||
| try: | try: | ||||
| branch_name = branch["name"] | branch_name = branch["name"] | ||||
| revision_id = branch["revision"] | revision_id = branch["revision"] | ||||
| root_sha1_git = branch["directory"] | root_directory = branch["directory"] | ||||
| except Exception as exc: | except Exception as exc: | ||||
| sentry_sdk.capture_exception(exc) | sentry_sdk.capture_exception(exc) | ||||
| _branch_not_found( | _branch_not_found( | ||||
| "branch", | "branch", | ||||
| branch_name, | branch_name, | ||||
| snapshot_id, | snapshot_id, | ||||
| snapshot_sizes, | snapshot_sizes, | ||||
| origin_info, | origin_info, | ||||
| timestamp, | timestamp, | ||||
| visit_id, | visit_id, | ||||
| ) | ) | ||||
| for b in branches: | for b in branches: | ||||
| branch_url_args = dict(url_args) | branch_url_args = dict(url_args) | ||||
| branch_query_params = dict(query_params) | branch_query_params = dict(query_params) | ||||
| if "release" in branch_query_params: | branch_query_params.pop("release", None) | ||||
| del branch_query_params["release"] | if b["name"] != b["revision"]: | ||||
| branch_query_params.pop("revision", None) | |||||
| branch_query_params["branch"] = b["name"] | branch_query_params["branch"] = b["name"] | ||||
| if path: | if path: | ||||
| b["path"] = path | |||||
| branch_url_args["path"] = path | branch_url_args["path"] = path | ||||
| b["url"] = reverse( | b["url"] = reverse( | ||||
| browse_view_name, url_args=branch_url_args, query_params=branch_query_params | browse_view_name, url_args=branch_url_args, query_params=branch_query_params | ||||
| ) | ) | ||||
| for r in releases: | for r in releases: | ||||
| release_url_args = dict(url_args) | release_url_args = dict(url_args) | ||||
| release_query_params = dict(query_params) | release_query_params = dict(query_params) | ||||
| if "branch" in release_query_params: | release_query_params.pop("branch", None) | ||||
| del release_query_params["branch"] | release_query_params.pop("revision", None) | ||||
| release_query_params["release"] = r["name"] | release_query_params["release"] = r["name"] | ||||
| if path: | if path: | ||||
| r["path"] = path | |||||
| release_url_args["path"] = path | release_url_args["path"] = path | ||||
| r["url"] = reverse( | r["url"] = reverse( | ||||
| browse_view_name, | browse_view_name, | ||||
| url_args=release_url_args, | url_args=release_url_args, | ||||
| query_params=release_query_params, | query_params=release_query_params, | ||||
| ) | ) | ||||
| snapshot_context["query_params"] = query_params | return SnapshotContext( | ||||
| snapshot_context["root_sha1_git"] = root_sha1_git | branch=branch_name, | ||||
| snapshot_context["revision_id"] = revision_id | branches=branches, | ||||
| snapshot_context["branch"] = branch_name | branches_url=branches_url, | ||||
| snapshot_context["release"] = release_name | is_empty=is_empty, | ||||
| snapshot_context["release_id"] = release_id | origin_info=origin_info, | ||||
| origin_visits_url=origin_visits_url, | |||||
| release=release_name, | |||||
| release_id=release_id, | |||||
| query_params=query_params, | |||||
| releases=releases, | |||||
| releases_url=releases_url, | |||||
| revision_id=revision_id, | |||||
| root_directory=root_directory, | |||||
| snapshot_id=snapshot_id, | |||||
| snapshot_sizes=snapshot_sizes, | |||||
| snapshot_swhid=swh_snp_id, | |||||
| url_args=url_args, | |||||
| visit_info=visit_info, | |||||
| ) | |||||
| return snapshot_context | |||||
| def _build_breadcrumbs(snapshot_context: SnapshotContext, path: str): | |||||
| origin_info = snapshot_context["origin_info"] | |||||
| url_args = snapshot_context["url_args"] | |||||
| query_params = snapshot_context["query_params"] | |||||
| root_directory = snapshot_context["root_directory"] | |||||
| path_info = gen_path_info(path) | |||||
| if origin_info: | |||||
| browse_view_name = "browse-origin-directory" | |||||
| else: | |||||
| browse_view_name = "browse-snapshot-directory" | |||||
| breadcrumbs = [] | |||||
| if root_directory: | |||||
| breadcrumbs.append( | |||||
| { | |||||
| "name": root_directory[:7], | |||||
| "url": reverse( | |||||
| browse_view_name, url_args=url_args, query_params=query_params | |||||
| ), | |||||
| } | |||||
| ) | |||||
| for pi in path_info: | |||||
| bc_url_args = dict(url_args) | |||||
| bc_url_args["path"] = pi["path"] | |||||
| breadcrumbs.append( | |||||
| { | |||||
| "name": pi["name"], | |||||
| "url": reverse( | |||||
| browse_view_name, url_args=bc_url_args, query_params=query_params | |||||
| ), | |||||
| } | |||||
| ) | |||||
| return breadcrumbs | |||||
| def browse_snapshot_directory( | def browse_snapshot_directory( | ||||
| request, snapshot_id=None, origin_url=None, timestamp=None, path=None | request, snapshot_id=None, origin_url=None, timestamp=None, path=None | ||||
| ): | ): | ||||
| """ | """ | ||||
| Django view implementation for browsing a directory in a snapshot context. | Django view implementation for browsing a directory in a snapshot context. | ||||
| """ | """ | ||||
| try: | try: | ||||
| snapshot_context = _process_snapshot_request( | snapshot_context = get_snapshot_context( | ||||
| request, | snapshot_id=snapshot_id, | ||||
| snapshot_id, | origin_url=origin_url, | ||||
| origin_url, | timestamp=timestamp, | ||||
| timestamp, | visit_id=request.GET.get("visit_id"), | ||||
| path, | path=path, | ||||
| browse_context="directory", | browse_context="directory", | ||||
| branch_name=request.GET.get("branch"), | |||||
| release_name=request.GET.get("release"), | |||||
ardumont: It's retrieving the GET query parameters, right?
(i forgot ;) | |||||
Done Inline ActionsYes it is ;-) anlambert: Yes it is ;-) | |||||
| revision_id=request.GET.get("revision"), | |||||
| ) | ) | ||||
| root_sha1_git = snapshot_context["root_sha1_git"] | root_directory = snapshot_context["root_directory"] | ||||
| sha1_git = root_sha1_git | sha1_git = root_directory | ||||
| if root_sha1_git and path: | if root_directory and path: | ||||
| dir_info = service.lookup_directory_with_path(root_sha1_git, path) | dir_info = service.lookup_directory_with_path(root_directory, path) | ||||
| sha1_git = dir_info["target"] | sha1_git = dir_info["target"] | ||||
| dirs = [] | dirs = [] | ||||
| files = [] | files = [] | ||||
| if sha1_git: | if sha1_git: | ||||
| dirs, files = get_directory_entries(sha1_git) | dirs, files = get_directory_entries(sha1_git) | ||||
| except Exception as exc: | except Exception as exc: | ||||
| return handle_view_exception(request, exc) | return handle_view_exception(request, exc) | ||||
| swh_type = snapshot_context["swh_type"] | |||||
| origin_info = snapshot_context["origin_info"] | origin_info = snapshot_context["origin_info"] | ||||
| visit_info = snapshot_context["visit_info"] | visit_info = snapshot_context["visit_info"] | ||||
| url_args = snapshot_context["url_args"] | url_args = snapshot_context["url_args"] | ||||
| query_params = snapshot_context["query_params"] | query_params = snapshot_context["query_params"] | ||||
| revision_id = snapshot_context["revision_id"] | revision_id = snapshot_context["revision_id"] | ||||
| snapshot_id = snapshot_context["snapshot_id"] | snapshot_id = snapshot_context["snapshot_id"] | ||||
| path_info = gen_path_info(path) | if origin_info: | ||||
| browse_view_name = "browse-origin-directory" | |||||
| browse_view_name = "browse-" + swh_type + "-directory" | else: | ||||
| browse_view_name = "browse-snapshot-directory" | |||||
| breadcrumbs = [] | breadcrumbs = _build_breadcrumbs(snapshot_context, path) | ||||
| if root_sha1_git: | |||||
| breadcrumbs.append( | |||||
| { | |||||
| "name": root_sha1_git[:7], | |||||
| "url": reverse( | |||||
| browse_view_name, url_args=url_args, query_params=query_params | |||||
| ), | |||||
| } | |||||
| ) | |||||
| for pi in path_info: | |||||
| bc_url_args = dict(url_args) | |||||
| bc_url_args["path"] = pi["path"] | |||||
| breadcrumbs.append( | |||||
| { | |||||
| "name": pi["name"], | |||||
| "url": reverse( | |||||
| browse_view_name, url_args=bc_url_args, query_params=query_params | |||||
| ), | |||||
| } | |||||
| ) | |||||
| path = "" if path is None else (path + "/") | path = "" if path is None else (path + "/") | ||||
| for d in dirs: | for d in dirs: | ||||
| if d["type"] == "rev": | if d["type"] == "rev": | ||||
| d["url"] = reverse("browse-revision", url_args={"sha1_git": d["target"]}) | d["url"] = reverse("browse-revision", url_args={"sha1_git": d["target"]}) | ||||
| else: | else: | ||||
| bc_url_args = dict(url_args) | bc_url_args = dict(url_args) | ||||
| bc_url_args["path"] = path + d["name"] | bc_url_args["path"] = path + d["name"] | ||||
| d["url"] = reverse( | d["url"] = reverse( | ||||
| browse_view_name, url_args=bc_url_args, query_params=query_params | browse_view_name, url_args=bc_url_args, query_params=query_params | ||||
| ) | ) | ||||
| sum_file_sizes = 0 | sum_file_sizes = 0 | ||||
| readmes = {} | readmes = {} | ||||
| browse_view_name = "browse-" + swh_type + "-content" | if origin_info: | ||||
| browse_view_name = "browse-origin-content" | |||||
| else: | |||||
| browse_view_name = "browse-snapshot-content" | |||||
| for f in files: | for f in files: | ||||
| bc_url_args = dict(url_args) | bc_url_args = dict(url_args) | ||||
| bc_url_args["path"] = path + f["name"] | bc_url_args["path"] = path + f["name"] | ||||
| f["url"] = reverse( | f["url"] = reverse( | ||||
| browse_view_name, url_args=bc_url_args, query_params=query_params | browse_view_name, url_args=bc_url_args, query_params=query_params | ||||
| ) | ) | ||||
| if f["length"] is not None: | if f["length"] is not None: | ||||
| sum_file_sizes += f["length"] | sum_file_sizes += f["length"] | ||||
| f["length"] = filesizeformat(f["length"]) | f["length"] = filesizeformat(f["length"]) | ||||
| if f["name"].lower().startswith("readme"): | if f["name"].lower().startswith("readme"): | ||||
| readmes[f["name"]] = f["checksums"]["sha1"] | readmes[f["name"]] = f["checksums"]["sha1"] | ||||
| readme_name, readme_url, readme_html = get_readme_to_display(readmes) | readme_name, readme_url, readme_html = get_readme_to_display(readmes) | ||||
| browse_view_name = "browse-" + swh_type + "-log" | if origin_info: | ||||
| browse_view_name = "browse-origin-log" | |||||
| else: | |||||
| browse_view_name = "browse-snapshot-log" | |||||
| history_url = None | history_url = None | ||||
| if snapshot_id != _empty_snapshot_id: | if snapshot_id != _empty_snapshot_id: | ||||
| history_url = reverse( | history_url = reverse( | ||||
| browse_view_name, url_args=url_args, query_params=query_params | browse_view_name, url_args=url_args, query_params=query_params | ||||
| ) | ) | ||||
| nb_files = None | nb_files = None | ||||
| nb_dirs = None | nb_dirs = None | ||||
| dir_path = None | dir_path = None | ||||
| if root_sha1_git: | if root_directory: | ||||
| nb_files = len(files) | nb_files = len(files) | ||||
| nb_dirs = len(dirs) | nb_dirs = len(dirs) | ||||
| sum_file_sizes = filesizeformat(sum_file_sizes) | sum_file_sizes = filesizeformat(sum_file_sizes) | ||||
| dir_path = "/" + path | dir_path = "/" + path | ||||
| browse_dir_link = gen_directory_link(sha1_git) | browse_dir_link = gen_directory_link(sha1_git) | ||||
| browse_rev_link = gen_revision_link(revision_id) | browse_rev_link = gen_revision_link(revision_id) | ||||
| ▲ Show 20 Lines • Show All 61 Lines • ▼ Show 20 Lines | return render( | ||||
| request, | request, | ||||
| "browse/directory.html", | "browse/directory.html", | ||||
| { | { | ||||
| "heading": heading, | "heading": heading, | ||||
| "swh_object_name": "Directory", | "swh_object_name": "Directory", | ||||
| "swh_object_metadata": dir_metadata, | "swh_object_metadata": dir_metadata, | ||||
| "dirs": dirs, | "dirs": dirs, | ||||
| "files": files, | "files": files, | ||||
| "breadcrumbs": breadcrumbs if root_sha1_git else [], | "breadcrumbs": breadcrumbs if root_directory else [], | ||||
| "top_right_link": top_right_link, | "top_right_link": top_right_link, | ||||
| "readme_name": readme_name, | "readme_name": readme_name, | ||||
| "readme_url": readme_url, | "readme_url": readme_url, | ||||
| "readme_html": readme_html, | "readme_html": readme_html, | ||||
| "snapshot_context": snapshot_context, | "snapshot_context": snapshot_context, | ||||
| "vault_cooking": vault_cooking, | "vault_cooking": vault_cooking, | ||||
| "show_actions_menu": True, | "show_actions_menu": True, | ||||
| "swh_ids": swh_ids, | "swh_ids": swh_ids, | ||||
| Show All 9 Lines | def browse_snapshot_content( | ||||
| path=None, | path=None, | ||||
| selected_language=None, | selected_language=None, | ||||
| ): | ): | ||||
| """ | """ | ||||
| Django view implementation for browsing a content in a snapshot context. | Django view implementation for browsing a content in a snapshot context. | ||||
| """ | """ | ||||
| try: | try: | ||||
| snapshot_context = _process_snapshot_request( | snapshot_context = get_snapshot_context( | ||||
| request, snapshot_id, origin_url, timestamp, path, browse_context="content" | snapshot_id=snapshot_id, | ||||
| origin_url=origin_url, | |||||
| timestamp=timestamp, | |||||
| visit_id=request.GET.get("visit_id"), | |||||
| path=path, | |||||
| browse_context="content", | |||||
| branch_name=request.GET.get("branch"), | |||||
| release_name=request.GET.get("release"), | |||||
| revision_id=request.GET.get("revision"), | |||||
| ) | ) | ||||
| root_sha1_git = snapshot_context["root_sha1_git"] | root_directory = snapshot_context["root_directory"] | ||||
| sha1_git = None | sha1_git = None | ||||
| query_string = None | query_string = None | ||||
| content_data = None | content_data = None | ||||
| directory_id = None | directory_id = None | ||||
| split_path = path.split("/") | split_path = path.split("/") | ||||
| filename = split_path[-1] | filename = split_path[-1] | ||||
| filepath = path[: -len(filename)] | filepath = path[: -len(filename)] | ||||
| if root_sha1_git: | if root_directory: | ||||
| content_info = service.lookup_directory_with_path(root_sha1_git, path) | content_info = service.lookup_directory_with_path(root_directory, path) | ||||
| sha1_git = content_info["target"] | sha1_git = content_info["target"] | ||||
| query_string = "sha1_git:" + sha1_git | query_string = "sha1_git:" + sha1_git | ||||
| content_data = request_content(query_string, raise_if_unavailable=False) | content_data = request_content(query_string, raise_if_unavailable=False) | ||||
| if filepath: | if filepath: | ||||
| dir_info = service.lookup_directory_with_path(root_sha1_git, filepath) | dir_info = service.lookup_directory_with_path(root_directory, filepath) | ||||
| directory_id = dir_info["target"] | directory_id = dir_info["target"] | ||||
| else: | else: | ||||
| directory_id = root_sha1_git | directory_id = root_directory | ||||
| except Exception as exc: | except Exception as exc: | ||||
| return handle_view_exception(request, exc) | return handle_view_exception(request, exc) | ||||
| swh_type = snapshot_context["swh_type"] | |||||
| url_args = snapshot_context["url_args"] | |||||
| query_params = snapshot_context["query_params"] | |||||
| revision_id = snapshot_context["revision_id"] | revision_id = snapshot_context["revision_id"] | ||||
| origin_info = snapshot_context["origin_info"] | origin_info = snapshot_context["origin_info"] | ||||
| visit_info = snapshot_context["visit_info"] | visit_info = snapshot_context["visit_info"] | ||||
| snapshot_id = snapshot_context["snapshot_id"] | snapshot_id = snapshot_context["snapshot_id"] | ||||
| content = None | content = None | ||||
| language = None | language = None | ||||
| mimetype = None | mimetype = None | ||||
| Show All 9 Lines | ): | ||||
| if selected_language is not None: | if selected_language is not None: | ||||
| language = selected_language | language = selected_language | ||||
| available_languages = None | available_languages = None | ||||
| if mimetype and "text/" in mimetype: | if mimetype and "text/" in mimetype: | ||||
| available_languages = highlightjs.get_supported_languages() | available_languages = highlightjs.get_supported_languages() | ||||
| browse_view_name = "browse-" + swh_type + "-directory" | breadcrumbs = _build_breadcrumbs(snapshot_context, filepath) | ||||
| breadcrumbs = [] | |||||
| path_info = gen_path_info(filepath) | |||||
| if root_sha1_git: | |||||
| breadcrumbs.append( | |||||
| { | |||||
| "name": root_sha1_git[:7], | |||||
| "url": reverse( | |||||
| browse_view_name, url_args=url_args, query_params=query_params | |||||
| ), | |||||
| } | |||||
| ) | |||||
| for pi in path_info: | |||||
| bc_url_args = dict(url_args) | |||||
| bc_url_args["path"] = pi["path"] | |||||
| breadcrumbs.append( | |||||
| { | |||||
| "name": pi["name"], | |||||
| "url": reverse( | |||||
| browse_view_name, url_args=bc_url_args, query_params=query_params | |||||
| ), | |||||
| } | |||||
| ) | |||||
| breadcrumbs.append({"name": filename, "url": None}) | breadcrumbs.append({"name": filename, "url": None}) | ||||
| browse_content_link = gen_content_link(sha1_git) | browse_content_link = gen_content_link(sha1_git) | ||||
| content_raw_url = None | content_raw_url = None | ||||
| if query_string: | if query_string: | ||||
| content_raw_url = reverse( | content_raw_url = reverse( | ||||
| ▲ Show 20 Lines • Show All 88 Lines • ▼ Show 20 Lines | return render( | ||||
| "swh_object_name": "Content", | "swh_object_name": "Content", | ||||
| "swh_object_metadata": content_metadata, | "swh_object_metadata": content_metadata, | ||||
| "content": content, | "content": content, | ||||
| "content_size": content_size, | "content_size": content_size, | ||||
| "max_content_size": content_display_max_size, | "max_content_size": content_display_max_size, | ||||
| "mimetype": mimetype, | "mimetype": mimetype, | ||||
| "language": language, | "language": language, | ||||
| "available_languages": available_languages, | "available_languages": available_languages, | ||||
| "breadcrumbs": breadcrumbs if root_sha1_git else [], | "breadcrumbs": breadcrumbs if root_directory else [], | ||||
| "top_right_link": top_right_link, | "top_right_link": top_right_link, | ||||
| "snapshot_context": snapshot_context, | "snapshot_context": snapshot_context, | ||||
| "vault_cooking": None, | "vault_cooking": None, | ||||
| "show_actions_menu": True, | "show_actions_menu": True, | ||||
| "swh_ids": swh_ids, | "swh_ids": swh_ids, | ||||
| "error_code": error_code, | "error_code": error_code, | ||||
| "error_message": error_message, | "error_message": error_message, | ||||
| "error_description": error_description, | "error_description": error_description, | ||||
| }, | }, | ||||
| status=error_code, | status=error_code, | ||||
| ) | ) | ||||
| PER_PAGE = 100 | PER_PAGE = 100 | ||||
| def browse_snapshot_log(request, snapshot_id=None, origin_url=None, timestamp=None): | def browse_snapshot_log(request, snapshot_id=None, origin_url=None, timestamp=None): | ||||
| """ | """ | ||||
| Django view implementation for browsing a revision history in a | Django view implementation for browsing a revision history in a | ||||
| snapshot context. | snapshot context. | ||||
| """ | """ | ||||
| try: | try: | ||||
| snapshot_context = _process_snapshot_request( | snapshot_context = get_snapshot_context( | ||||
| request, snapshot_id, origin_url, timestamp, browse_context="log" | snapshot_id=snapshot_id, | ||||
| origin_url=origin_url, | |||||
| timestamp=timestamp, | |||||
| visit_id=request.GET.get("visit_id"), | |||||
| browse_context="log", | |||||
| branch_name=request.GET.get("branch"), | |||||
| release_name=request.GET.get("release"), | |||||
| revision_id=request.GET.get("revision"), | |||||
| ) | ) | ||||
| revision_id = snapshot_context["revision_id"] | revision_id = snapshot_context["revision_id"] | ||||
| per_page = int(request.GET.get("per_page", PER_PAGE)) | per_page = int(request.GET.get("per_page", PER_PAGE)) | ||||
| offset = int(request.GET.get("offset", 0)) | offset = int(request.GET.get("offset", 0)) | ||||
| revs_ordering = request.GET.get("revs_ordering", "committer_date") | revs_ordering = request.GET.get("revs_ordering", "committer_date") | ||||
| session_key = "rev_%s_log_ordering_%s" % (revision_id, revs_ordering) | session_key = "rev_%s_log_ordering_%s" % (revision_id, revs_ordering) | ||||
| Show All 20 Lines | try: | ||||
| request.session[session_key] = { | request.session[session_key] = { | ||||
| "rev_log": rev_log, | "rev_log": rev_log, | ||||
| "revs_walker_state": revs_walker_state, | "revs_walker_state": revs_walker_state, | ||||
| } | } | ||||
| except Exception as exc: | except Exception as exc: | ||||
| return handle_view_exception(request, exc) | return handle_view_exception(request, exc) | ||||
| swh_type = snapshot_context["swh_type"] | |||||
| origin_info = snapshot_context["origin_info"] | origin_info = snapshot_context["origin_info"] | ||||
| visit_info = snapshot_context["visit_info"] | visit_info = snapshot_context["visit_info"] | ||||
| url_args = snapshot_context["url_args"] | url_args = snapshot_context["url_args"] | ||||
| query_params = snapshot_context["query_params"] | query_params = snapshot_context["query_params"] | ||||
| snapshot_id = snapshot_context["snapshot_id"] | snapshot_id = snapshot_context["snapshot_id"] | ||||
| query_params["per_page"] = per_page | query_params["per_page"] = per_page | ||||
| revs_ordering = request.GET.get("revs_ordering", "") | revs_ordering = request.GET.get("revs_ordering", "") | ||||
| query_params["revs_ordering"] = revs_ordering | query_params["revs_ordering"] = revs_ordering | ||||
| browse_view_name = "browse-" + swh_type + "-log" | if origin_info: | ||||
| browse_view_name = "browse-origin-log" | |||||
| else: | |||||
| browse_view_name = "browse-snapshot-log" | |||||
| prev_log_url = None | prev_log_url = None | ||||
| if len(rev_log) > offset + per_page: | if len(rev_log) > offset + per_page: | ||||
| query_params["offset"] = offset + per_page | query_params["offset"] = offset + per_page | ||||
| prev_log_url = reverse( | prev_log_url = reverse( | ||||
| browse_view_name, url_args=url_args, query_params=query_params | browse_view_name, url_args=url_args, query_params=query_params | ||||
| ) | ) | ||||
| ▲ Show 20 Lines • Show All 67 Lines • ▼ Show 20 Lines | |||||
| def browse_snapshot_branches( | def browse_snapshot_branches( | ||||
| request, snapshot_id=None, origin_url=None, timestamp=None | request, snapshot_id=None, origin_url=None, timestamp=None | ||||
| ): | ): | ||||
| """ | """ | ||||
| Django view implementation for browsing a list of branches in a snapshot | Django view implementation for browsing a list of branches in a snapshot | ||||
| context. | context. | ||||
| """ | """ | ||||
| try: | try: | ||||
| snapshot_context = _process_snapshot_request( | |||||
| request, snapshot_id, origin_url, timestamp | snapshot_context = get_snapshot_context( | ||||
| snapshot_id=snapshot_id, | |||||
| origin_url=origin_url, | |||||
| timestamp=timestamp, | |||||
| visit_id=request.GET.get("visit_id"), | |||||
| ) | ) | ||||
| branches_bc = request.GET.get("branches_breadcrumbs", "") | branches_bc = request.GET.get("branches_breadcrumbs", "") | ||||
| branches_bc = branches_bc.split(",") if branches_bc else [] | branches_bc = branches_bc.split(",") if branches_bc else [] | ||||
| branches_from = branches_bc[-1] if branches_bc else "" | branches_from = branches_bc[-1] if branches_bc else "" | ||||
| swh_type = snapshot_context["swh_type"] | |||||
| origin_info = snapshot_context["origin_info"] | origin_info = snapshot_context["origin_info"] | ||||
| url_args = snapshot_context["url_args"] | url_args = snapshot_context["url_args"] | ||||
| query_params = snapshot_context["query_params"] | query_params = snapshot_context["query_params"] | ||||
| browse_view_name = "browse-" + swh_type + "-directory" | if origin_info: | ||||
| browse_view_name = "browse-origin-directory" | |||||
| else: | |||||
| browse_view_name = "browse-snapshot-directory" | |||||
| snapshot = service.lookup_snapshot( | snapshot = service.lookup_snapshot( | ||||
| snapshot_context["snapshot_id"], | snapshot_context["snapshot_id"], | ||||
| branches_from, | branches_from, | ||||
| PER_PAGE + 1, | PER_PAGE + 1, | ||||
| target_types=["revision", "alias"], | target_types=["revision", "alias"], | ||||
| ) | ) | ||||
| Show All 18 Lines | for branch in displayed_branches: | ||||
| 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"] | ||||
| branch["revision_url"] = revision_url | branch["revision_url"] = revision_url | ||||
| branch["directory_url"] = directory_url | branch["directory_url"] = directory_url | ||||
| browse_view_name = "browse-" + swh_type + "-branches" | if origin_info: | ||||
| browse_view_name = "browse-origin-branches" | |||||
| else: | |||||
| browse_view_name = "browse-snapshot-branches" | |||||
| prev_branches_url = None | prev_branches_url = None | ||||
| next_branches_url = None | next_branches_url = None | ||||
| if branches_bc: | if branches_bc: | ||||
| query_params_prev = dict(query_params) | query_params_prev = dict(query_params) | ||||
| query_params_prev["branches_breadcrumbs"] = ",".join(branches_bc[:-1]) | query_params_prev["branches_breadcrumbs"] = ",".join(branches_bc[:-1]) | ||||
| Show All 40 Lines | |||||
| def browse_snapshot_releases( | def browse_snapshot_releases( | ||||
| request, snapshot_id=None, origin_url=None, timestamp=None | request, snapshot_id=None, origin_url=None, timestamp=None | ||||
| ): | ): | ||||
| """ | """ | ||||
| Django view implementation for browsing a list of releases in a snapshot | Django view implementation for browsing a list of releases in a snapshot | ||||
| context. | context. | ||||
| """ | """ | ||||
| try: | try: | ||||
| snapshot_context = _process_snapshot_request( | snapshot_context = get_snapshot_context( | ||||
| request, snapshot_id, origin_url, timestamp | snapshot_id=snapshot_id, | ||||
| origin_url=origin_url, | |||||
| timestamp=timestamp, | |||||
| visit_id=request.GET.get("visit_id"), | |||||
| ) | ) | ||||
| rel_bc = request.GET.get("releases_breadcrumbs", "") | rel_bc = request.GET.get("releases_breadcrumbs", "") | ||||
| rel_bc = rel_bc.split(",") if rel_bc else [] | rel_bc = rel_bc.split(",") if rel_bc else [] | ||||
| rel_from = rel_bc[-1] if rel_bc else "" | rel_from = rel_bc[-1] if rel_bc else "" | ||||
| swh_type = snapshot_context["swh_type"] | |||||
| origin_info = snapshot_context["origin_info"] | origin_info = snapshot_context["origin_info"] | ||||
| url_args = snapshot_context["url_args"] | url_args = snapshot_context["url_args"] | ||||
| query_params = snapshot_context["query_params"] | query_params = snapshot_context["query_params"] | ||||
| snapshot = service.lookup_snapshot( | snapshot = service.lookup_snapshot( | ||||
| snapshot_context["snapshot_id"], | snapshot_context["snapshot_id"], | ||||
| rel_from, | rel_from, | ||||
| PER_PAGE + 1, | PER_PAGE + 1, | ||||
| Show All 40 Lines | for release in displayed_releases: | ||||
| "browse-release", | "browse-release", | ||||
| url_args={"sha1_git": release["target"]}, | url_args={"sha1_git": release["target"]}, | ||||
| query_params=query_params_tgt, | query_params=query_params_tgt, | ||||
| ) | ) | ||||
| release["release_url"] = release_url | release["release_url"] = release_url | ||||
| release["target_url"] = target_url | release["target_url"] = target_url | ||||
| browse_view_name = "browse-" + swh_type + "-releases" | if origin_info: | ||||
| browse_view_name = "browse-origin-releases" | |||||
| else: | |||||
| browse_view_name = "browse-snapshot-releases" | |||||
| prev_releases_url = None | prev_releases_url = None | ||||
| next_releases_url = None | next_releases_url = None | ||||
| if rel_bc: | if rel_bc: | ||||
| query_params_prev = dict(query_params) | query_params_prev = dict(query_params) | ||||
| query_params_prev["releases_breadcrumbs"] = ",".join(rel_bc[:-1]) | query_params_prev["releases_breadcrumbs"] = ",".join(rel_bc[:-1]) | ||||
| ▲ Show 20 Lines • Show All 42 Lines • Show Last 20 Lines | |||||
It's retrieving the GET query parameters, right?
(i forgot ;)