Differential D8889 Diff 32077 swh/web/save_code_now/api_views.py

Changeset View

Standalone View

swh/web/save_code_now/api_views.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

import os		import os
		import textwrap
from typing import Optional, cast		from typing import Optional, cast

		from django.conf import settings
from rest_framework.request import Request		from rest_framework.request import Request

from swh.web.api.apidoc import api_doc, format_docstring		from swh.web.api.apidoc import api_doc, format_docstring
from swh.web.api.apiurls import APIUrls, api_route		from swh.web.api.apiurls import APIUrls, api_route
from swh.web.auth.utils import (		from swh.web.auth.utils import (
API_SAVE_ORIGIN_PERMISSION,		API_SAVE_ORIGIN_PERMISSION,
SWH_AMBASSADOR_PERMISSION,		SWH_AMBASSADOR_PERMISSION,
privileged_user,		privileged_user,
Show All 11 Lines	if os.environ.get("DJANGO_SETTINGS_MODULE") != "swh.web.settings.tests":
visit_types = sorted(get_savable_visit_types())		visit_types = sorted(get_savable_visit_types())
docstring = ""		docstring = ""
for visit_type in visit_types[:-1]:		for visit_type in visit_types[:-1]:
docstring += f"{visit_type}, "		docstring += f"{visit_type}, "
docstring += f"and {visit_types[-1]}"		docstring += f"and {visit_types[-1]}"
return docstring		return docstring


		def _webhook_info_doc() -> str:
		docstring = ""
		if "swh.web.save_origin_webhooks" in settings.SWH_DJANGO_APPS:
		docstring = textwrap.dedent(
		"""
		:>json boolean from_webhook: indicates if the save request was created
		from a popular forge webhook receiver
		(see :http:post:`/api/1/origin/save/webhook/github/` for instance)
		:>json string webhook_origin: indicates which forge type sent the webhook,
		currently the supported types are:
		"""
		)
		# instantiate webhook receivers
		from swh.web.save_origin_webhooks import urls # noqa
		from swh.web.save_origin_webhooks.generic_receiver import SUPPORTED_FORGE_TYPES

		webhook_forge_types = sorted(list(SUPPORTED_FORGE_TYPES))
		for visit_type in webhook_forge_types[:-1]:
		docstring += f"{visit_type}, "
		docstring += f"and {webhook_forge_types[-1]}"
		return docstring


save_code_now_api_urls = APIUrls()		save_code_now_api_urls = APIUrls()


@api_route(		@api_route(
r"/origin/save/(?P<visit_type>.+)/url/(?P<origin_url>.+)/",		r"/origin/save/(?P<visit_type>.+)/url/(?P<origin_url>.+)/",
"api-1-save-origin",		"api-1-save-origin",
methods=["GET", "POST"],		methods=["GET", "POST"],
throttle_scope="swh_save_origin",		throttle_scope="swh_save_origin",
never_cache=True,		never_cache=True,
api_urls=save_code_now_api_urls,		api_urls=save_code_now_api_urls,
)		)
@api_doc("/origin/save/", category="Request archival")		@api_doc("/origin/save/", category="Request archival")
@format_docstring(visit_types=_savable_visit_types())		@format_docstring(
		visit_types=_savable_visit_types(), webhook_info_doc=_webhook_info_doc()
		)
def api_save_origin(request: Request, visit_type: str, origin_url: str):		def api_save_origin(request: Request, visit_type: str, origin_url: str):
"""		"""
.. http:get:: /api/1/origin/save/(visit_type)/url/(origin_url)/		.. http:get:: /api/1/origin/save/(visit_type)/url/(origin_url)/
.. http:post:: /api/1/origin/save/(visit_type)/url/(origin_url)/		.. http:post:: /api/1/origin/save/(visit_type)/url/(origin_url)/

Request the saving of a software origin into the archive		Request the saving of a software origin into the archive
or check the status of previously created save requests.		or check the status of previously created save requests.

Show All 40 Lines	.. http:post:: /api/1/origin/save/(visit_type)/url/(origin_url)/
succeeded or failed		succeeded or failed
:>json string visit_date: the date (in iso format) of the visit if a visit		:>json string visit_date: the date (in iso format) of the visit if a visit
occurred, null otherwise.		occurred, null otherwise.
:>json string visit_status: the status of the visit, either full,		:>json string visit_status: the status of the visit, either full,
partial, not_found or failed if a visit occurred, null		partial, not_found or failed if a visit occurred, null
otherwise.		otherwise.
:>json string note: optional note giving details about the save request,		:>json string note: optional note giving details about the save request,
for instance why it has been rejected		for instance why it has been rejected
		{webhook_info_doc}

:statuscode 200: no error		:statuscode 200: no error
:statuscode 400: an invalid visit type or origin url has been provided		:statuscode 400: an invalid visit type or origin url has been provided
:statuscode 403: the provided origin url is blacklisted		:statuscode 403: the provided origin url is blacklisted
:statuscode 404: no save requests have been found for a given origin		:statuscode 404: no save requests have been found for a given origin

"""		"""

		def _cleanup_sor_data(sor):
		del sor["id"]
		if "swh.web.save_origin_webhooks" not in settings.SWH_DJANGO_APPS:
		del sor["from_webhook"]
		del sor["webhook_origin"]
		return sor

data = request.data or {}		data = request.data or {}
if request.method == "POST":		if request.method == "POST":
sor = create_save_origin_request(		sor = create_save_origin_request(
visit_type,		visit_type,
origin_url,		origin_url,
privileged_user(		privileged_user(
request,		request,
permissions=[SWH_AMBASSADOR_PERMISSION, API_SAVE_ORIGIN_PERMISSION],		permissions=[SWH_AMBASSADOR_PERMISSION, API_SAVE_ORIGIN_PERMISSION],
),		),
user_id=cast(Optional[int], request.user.id),		user_id=cast(Optional[int], request.user.id),
**data,		**data,
)		)
del sor["id"]		return _cleanup_sor_data(sor)
return sor
else:		else:
sors = get_save_origin_requests(visit_type, origin_url)		sors = get_save_origin_requests(visit_type, origin_url)
for sor in sors:		return [_cleanup_sor_data(sor) for sor in sors]
del sor["id"]
return sors