Changeset View
Changeset View
Standalone View
Standalone View
swh/web/api/views/utils.py
| # Copyright (C) 2015-2019 The Software Heritage developers | # Copyright (C) 2015-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 | ||||
| from types import GeneratorType | from types import GeneratorType | ||||
| from typing import Any, Callable, Dict, Mapping, Optional | from typing import Any, Callable, Dict, List, Optional, Tuple, Union | ||||
| from typing_extensions import Protocol | |||||
| from django.http import HttpRequest | from django.http import HttpRequest | ||||
| from rest_framework.decorators import api_view | from rest_framework.decorators import api_view | ||||
| from rest_framework.request import Request | |||||
| from rest_framework.response import Response | from rest_framework.response import Response | ||||
| from swh.web.api.apiurls import APIUrls, api_route | from swh.web.api.apiurls import APIUrls, api_route | ||||
| from swh.web.common.exc import NotFoundExc | from swh.web.common.exc import NotFoundExc | ||||
| EnrichFunction = Callable[[Dict[str, str], Optional[HttpRequest]], Dict[str, str]] | |||||
| class EnrichFunction(Protocol): | EnrichFunctionSearchResult = Callable[ | ||||
| def __call__( | [Tuple[List[Dict[str, Any]], Optional[str]], Optional[HttpRequest]], | ||||
| self, input: Mapping[str, str], request: Optional[HttpRequest] | Tuple[List[Dict[str, Any]], Optional[str]], | ||||
| ) -> Dict[str, str]: | ] | ||||
| ... | |||||
| def api_lookup( | def api_lookup( | ||||
| lookup_fn: Callable[..., Any], | lookup_fn: Callable[..., Any], | ||||
| *args: Any, | *args: Any, | ||||
| notfound_msg: Optional[str] = "Object not found", | notfound_msg: Optional[str] = "Object not found", | ||||
| enrich_fn: Optional[EnrichFunction] = None, | enrich_fn: Optional[Union[EnrichFunction, EnrichFunctionSearchResult]] = None, | ||||
| request: Optional[HttpRequest] = None, | request: Optional[HttpRequest] = None, | ||||
| **kwargs: Any, | **kwargs: Any, | ||||
| ): | ): | ||||
| r""" | r""" | ||||
| Capture a redundant behavior of: | Capture a redundant behavior of: | ||||
| - looking up the backend with a criteria (be it an identifier or | - looking up the backend with a criteria (be it an identifier or | ||||
| checksum) passed to the function lookup_fn | checksum) passed to the function lookup_fn | ||||
| - if nothing is found, raise an NotFoundExc exception with error | - if nothing is found, raise an NotFoundExc exception with error | ||||
| Show All 25 Lines | def _enrich_fn_noop(x, request): | ||||
| return x | return x | ||||
| if enrich_fn is None: | if enrich_fn is None: | ||||
| enrich_fn = _enrich_fn_noop | enrich_fn = _enrich_fn_noop | ||||
| res = lookup_fn(*args, **kwargs) | res = lookup_fn(*args, **kwargs) | ||||
| if res is None: | if res is None: | ||||
| raise NotFoundExc(notfound_msg) | raise NotFoundExc(notfound_msg) | ||||
| if isinstance(res, (list, GeneratorType)) or type(res) == map: | if isinstance(res, (list, GeneratorType)) or type(res) == map: | ||||
| return [enrich_fn(x, request=request) for x in res] | return [enrich_fn(x, request) for x in res] | ||||
| return enrich_fn(res, request=request) | return enrich_fn(res, request) | ||||
| @api_view(["GET", "HEAD"]) | @api_view(["GET", "HEAD"]) | ||||
| def api_home(request): | def api_home(request: Request): | ||||
| return Response({}, template_name="api/api.html") | return Response({}, template_name="api/api.html") | ||||
| APIUrls.add_url_pattern(r"^$", api_home, view_name="api-1-homepage") | APIUrls.add_url_pattern(r"^$", api_home, view_name="api-1-homepage") | ||||
| @api_route(r"/", "api-1-endpoints") | @api_route(r"/", "api-1-endpoints") | ||||
| def api_endpoints(request): | def api_endpoints(request): | ||||
| """Display the list of opened api endpoints.""" | """Display the list of opened api endpoints.""" | ||||
| routes = APIUrls.get_app_endpoints().copy() | routes = APIUrls.get_app_endpoints().copy() | ||||
| for route, doc in routes.items(): | for route, doc in routes.items(): | ||||
| doc["doc_intro"] = doc["docstring"].split("\n\n")[0] | doc["doc_intro"] = doc["docstring"].split("\n\n")[0] | ||||
| # Return a list of routes with consistent ordering | # Return a list of routes with consistent ordering | ||||
| env = {"doc_routes": sorted(routes.items())} | env = {"doc_routes": sorted(routes.items())} | ||||
| return Response(env, template_name="api/endpoints.html") | return Response(env, template_name="api/endpoints.html") | ||||