Differential D4276 Diff 15122 swh/web/api/apiresponse.py

Changeset View

Standalone View

swh/web/api/apiresponse.py

# Copyright (C) 2017-2019 The Software Heritage developers		# Copyright (C) 2017-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

import json		import json
import traceback		import traceback
from typing import Any, Dict, Optional		from typing import Any, Dict, Optional

		import sentry_sdk

		from django.http import HttpResponse
		from django.shortcuts import render
from django.utils.html import escape		from django.utils.html import escape
		from rest_framework.exceptions import APIException
from rest_framework.request import Request		from rest_framework.request import Request
from rest_framework.response import Response		from rest_framework.response import Response
from rest_framework.utils.encoders import JSONEncoder		from rest_framework.utils.encoders import JSONEncoder

from swh.storage.exc import StorageAPIError, StorageDBError		from swh.storage.exc import StorageAPIError, StorageDBError
from swh.web.api import utils		from swh.web.api import utils
from swh.web.common.exc import BadInputExc, ForbiddenExc, LargePayloadExc, NotFoundExc		from swh.web.common.exc import BadInputExc, ForbiddenExc, LargePayloadExc, NotFoundExc
from swh.web.common.utils import gen_path_info, shorten_path		from swh.web.common.utils import gen_path_info, shorten_path
▲ Show 20 Lines • Show All 72 Lines • ▼ Show 20 Lines	def transform(rv: Dict[str, Any]) -> Dict[str, Any]:
return rv		return rv


def make_api_response(		def make_api_response(
request: Request,		request: Request,
data: Dict[str, Any],		data: Dict[str, Any],
doc_data: Optional[Dict[str, Any]] = None,		doc_data: Optional[Dict[str, Any]] = None,
options: Optional[Dict[str, Any]] = None,		options: Optional[Dict[str, Any]] = None,
) -> Response:		) -> HttpResponse:
"""Generates an API response based on the requested mimetype.		"""Generates an API response based on the requested mimetype.

Args:		Args:
request: a DRF Request object		request: a DRF Request object
data: raw data to return in the API response		data: raw data to return in the API response
doc_data: documentation data for HTML response		doc_data: documentation data for HTML response
options: optional data that can be used to generate the response		options: optional data that can be used to generate the response

Show All 10 Lines	) -> HttpResponse:
headers = {}		headers = {}
if "headers" in options:		if "headers" in options:
doc_data["headers_data"] = options["headers"]		doc_data["headers_data"] = options["headers"]
headers = options["headers"]		headers = options["headers"]

# get request status code		# get request status code
doc_data["status_code"] = options.get("status", 200)		doc_data["status_code"] = options.get("status", 200)

response_args = {
"status": doc_data["status_code"],
"headers": headers,
"content_type": request.accepted_media_type,
}

# when requesting HTML, typically when browsing the API through its		# when requesting HTML, typically when browsing the API through its
# documented views, we need to enrich the input data with documentation		# documented views, we need to enrich the input data with documentation
# related ones and inform DRF that we request HTML template rendering		# and render the apidoc HTML template
if request.accepted_media_type == "text/html":		if request.accepted_media_type == "text/html":
doc_data["response_data"] = data		doc_data["response_data"] = data
if data:		if data:
doc_data["response_data"] = json.dumps(		doc_data["response_data"] = json.dumps(
data, cls=JSONEncoder, sort_keys=True, indent=4, separators=(",", ": ")		data, cls=JSONEncoder, sort_keys=True, indent=4, separators=(",", ": ")
)		)

doc_data["heading"] = shorten_path(str(request.path))		doc_data["heading"] = shorten_path(str(request.path))

# generate breadcrumbs data		# generate breadcrumbs data
if "route" in doc_data:		if "route" in doc_data:
doc_data["endpoint_path"] = gen_path_info(doc_data["route"])		doc_data["endpoint_path"] = gen_path_info(doc_data["route"])
for i in range(len(doc_data["endpoint_path"]) - 1):		for i in range(len(doc_data["endpoint_path"]) - 1):
doc_data["endpoint_path"][i]["path"] += "/doc/"		doc_data["endpoint_path"][i]["path"] += "/doc/"
if not doc_data["noargs"]:		if not doc_data["noargs"]:
doc_data["endpoint_path"][-1]["path"] += "/doc/"		doc_data["endpoint_path"][-1]["path"] += "/doc/"

response_args["data"] = doc_data		return render(
response_args["template_name"] = "api/apidoc.html"		request, "api/apidoc.html", doc_data, status=doc_data["status_code"]
		)

# otherwise simply return the raw data and let DRF picks		# otherwise simply return the raw data and let DRF picks
# the correct renderer (JSON or YAML)		# the correct renderer (JSON or YAML)
else:		else:
response_args["data"] = data		return Response(
		data,
return Response(**response_args)		headers=headers,
		content_type=request.accepted_media_type,
		status=doc_data["status_code"],
		)


def error_response(		def error_response(
request: Request, error: Exception, doc_data: Dict[str, Any]		request: Request, exception: Exception, doc_data: Dict[str, Any]
) -> Response:		) -> HttpResponse:
"""Private function to create a custom error response.		"""Private function to create a custom error response.

Args:		Args:
request: a DRF Request object		request: a DRF Request object
error: the exception that caused the error		error: the exception that caused the error
doc_data: documentation data for HTML response		doc_data: documentation data for HTML response

"""		"""
error_code = 500		error_code = 500
if isinstance(error, BadInputExc):		if isinstance(exception, BadInputExc):
error_code = 400		error_code = 400
elif isinstance(error, NotFoundExc):		elif isinstance(exception, NotFoundExc):
error_code = 404		error_code = 404
elif isinstance(error, ForbiddenExc):		elif isinstance(exception, ForbiddenExc):
error_code = 403		error_code = 403
elif isinstance(error, LargePayloadExc):		elif isinstance(exception, LargePayloadExc):
error_code = 413		error_code = 413
elif isinstance(error, StorageDBError):		elif isinstance(exception, StorageDBError):
error_code = 503		error_code = 503
elif isinstance(error, StorageAPIError):		elif isinstance(exception, StorageAPIError):
error_code = 503		error_code = 503
		elif isinstance(exception, APIException):
		error_code = exception.status_code

error_opts = {"status": error_code}		error_opts = {"status": error_code}
error_data = {		error_data = {
"exception": error.__class__.__name__,		"exception": exception.__class__.__name__,
"reason": str(error),		"reason": str(exception),
}		}

if request.accepted_media_type == "text/html":		if request.accepted_media_type == "text/html":
error_data["reason"] = escape(error_data["reason"])		error_data["reason"] = escape(error_data["reason"])

if get_config()["debug"]:		if get_config()["debug"]:
error_data["traceback"] = traceback.format_exc()		error_data["traceback"] = traceback.format_exc()

return make_api_response(request, error_data, doc_data, options=error_opts)		return make_api_response(request, error_data, doc_data, options=error_opts)


		def error_response_handler(
		exc: Exception, context: Dict[str, Any]
		) -> Optional[HttpResponse]:
		"""Custom DRF exception handler used to generate API error responses.
		"""
		sentry_sdk.capture_exception(exc)
		doc_data = getattr(exc, "doc_data") if hasattr(exc, "doc_data") else None
		vlorentzUnsubmitted Not Done Inline Actions equivalent to: `doc_data = getattr(exc, "doc_data", None)` vlorentz: equivalent to: `doc_data = getattr(exc, "doc_data", None)`
		return error_response(context["request"], exc, doc_data)