Source code for swh.web.api.views.graph

# Copyright (C) 2020-2022  The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information

from distutils.util import strtobool
import json
from typing import Dict, Iterator, Union
from urllib.parse import unquote, urlparse, urlunparse

import requests

from django.http import QueryDict
from django.http.response import StreamingHttpResponse
from rest_framework.decorators import renderer_classes
from rest_framework.renderers import JSONRenderer
from rest_framework.request import Request
from rest_framework.response import Response

from swh.model.hashutil import hash_to_hex
from swh.model.model import Sha1Git
from swh.model.swhids import ExtendedObjectType, ExtendedSWHID
from swh.web.api.apidoc import api_doc
from swh.web.api.apiurls import api_route
from swh.web.api.renderers import PlainTextRenderer
from swh.web.config import SWH_WEB_INTERNAL_SERVER_NAMES, get_config
from swh.web.utils import archive

API_GRAPH_PERM = "swh.web.api.graph"

def _resolve_origin_swhid(swhid: str, origin_urls: Dict[Sha1Git, str]) -> str:
    Resolve origin url from its swhid sha1 representation.
    parsed_swhid = ExtendedSWHID.from_string(swhid)
    if parsed_swhid.object_type == ExtendedObjectType.ORIGIN:
        if parsed_swhid.object_id in origin_urls:
            return origin_urls[parsed_swhid.object_id]
            origin_info = list(
            assert origin_info is not None
            origin_urls[parsed_swhid.object_id] = origin_info["url"]
            return origin_info["url"]
        return swhid

def _resolve_origin_swhids_in_graph_response(
    response: requests.Response,
) -> Iterator[bytes]:
    Resolve origin urls from their swhid sha1 representations in graph service
    content_type = response.headers["Content-Type"]
    origin_urls: Dict[Sha1Git, str] = {}
    if content_type == "application/x-ndjson":
        for line in response.iter_lines():
            swhids = json.loads(line.decode("utf-8"))
            processed_line = []
            for swhid in swhids:
                processed_line.append(_resolve_origin_swhid(swhid, origin_urls))
            yield (json.dumps(processed_line) + "\n").encode()
    elif content_type == "text/plain":
        for line in response.iter_lines():
            if not line:
            processed_line = []
            swhids = line.decode("utf-8").split(" ")
            for swhid in swhids:
                processed_line.append(_resolve_origin_swhid(swhid, origin_urls))
            yield (" ".join(processed_line) + "\n").encode()
        for line in response.iter_lines():
            yield line + b"\n"

[docs] @api_route(r"/graph/", "api-1-graph-doc") @api_doc("/graph/", category="Miscellaneous") def api_graph(request: Request) -> None: """ .. http:get:: /api/1/graph/(graph_query)/ Provide fast access to the graph representation of the Software Heritage archive. That endpoint acts as a proxy for the `Software Heritage Graph service <>`_. It provides fast access to the `graph representation <>`_ of the Software Heritage archive. For more details please refer to the `Graph RPC API documentation <>`_. .. warning:: That endpoint is not publicly available and requires authentication and special user permission in order to be able to request it. :param string graph_query: query to forward to the Software Heritage Graph archive (see its `documentation <>`_) :query boolean resolve_origins: extra parameter defined by that proxy enabling to resolve origin urls from their sha1 representations :statuscode 200: no error :statuscode 400: an invalid graph query has been provided :statuscode 404: provided graph node cannot be found **Examples:** .. parsed-literal:: :swh_web_api:`graph/leaves/swh:1:dir:432d1b21c1256f7408a07c577b6974bbdbcc1323/` :swh_web_api:`graph/neighbors/swh:1:rev:f39d7d78b70e0f39facb1e4fab77ad3df5c52a35/` :swh_web_api:`graph/visit/nodes/swh:1:snp:40f9f177b8ab0b7b3d70ee14bbc8b214e2b2dcfc?direction=backward&resolve_origins=true` :swh_web_api:`graph/visit/edges/swh:1:snp:40f9f177b8ab0b7b3d70ee14bbc8b214e2b2dcfc?direction=backward&resolve_origins=true` """ return None
[docs] @api_route(r"/graph/(?P<graph_query>.+)/", "api-1-graph") @renderer_classes([JSONRenderer, PlainTextRenderer]) def api_graph_proxy( request: Request, graph_query: str ) -> Union[Response, StreamingHttpResponse]: if request.get_host() not in SWH_WEB_INTERNAL_SERVER_NAMES: if not bool(request.user and request.user.is_authenticated): return Response("Authentication credentials were not provided.", status=401) if not request.user.has_perm(API_GRAPH_PERM): return Response( "You do not have permission to perform this action.", status=403 ) graph_config = get_config()["graph"] graph_query = unquote(graph_query) graph_query_url = graph_config["server_url"] graph_query_url += graph_query parsed_url = urlparse(graph_query_url) query_dict = QueryDict(parsed_url.query, mutable=True) query_dict.update(request.GET) # clamp max_edges query parameter according to authentication if request.user.is_staff: max_edges = graph_config["max_edges"]["staff"] elif request.user.is_authenticated: max_edges = graph_config["max_edges"]["user"] else: max_edges = graph_config["max_edges"]["anonymous"] query_dict["max_edges"] = min( max_edges, int(query_dict.get("max_edges", max_edges + 1)) ) if query_dict: graph_query_url = urlunparse( parsed_url._replace(query=query_dict.urlencode(safe="/;:")) ) response = requests.get(graph_query_url, stream=True) if response.status_code != 200: return Response( response.content, status=response.status_code, content_type=response.headers["Content-Type"], ) # graph stats and counter endpoint responses are not streamed if response.headers.get("Transfer-Encoding") != "chunked": return Response( response.json(), status=response.status_code, content_type=response.headers["Content-Type"], ) # other endpoint responses are streamed else: resolve_origins = strtobool(request.GET.get("resolve_origins", "false")) if response.status_code == 200 and resolve_origins: response_stream = _resolve_origin_swhids_in_graph_response(response) else: response_stream = map(lambda line: line + b"\n", response.iter_lines()) return StreamingHttpResponse( response_stream, status=response.status_code, content_type=response.headers["Content-Type"], )