Changeset View
Changeset View
Standalone View
Standalone View
swh/deposit/api/edit_media.py
# Copyright (C) 2017-2020 The Software Heritage developers | # Copyright (C) 2017-2020 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 General Public License version 3, or any later version | # License: GNU 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 typing import Any, Dict, Optional, Tuple | from typing import Dict, Optional, Tuple | ||||
from rest_framework import status | from rest_framework import status | ||||
from ..config import CONT_FILE_IRI | from ..config import CONT_FILE_IRI | ||||
from ..errors import BAD_REQUEST, DepositError | from ..errors import BAD_REQUEST, DepositError | ||||
from ..parsers import SWHFileUploadTarParser, SWHFileUploadZipParser | from ..parsers import SWHFileUploadTarParser, SWHFileUploadZipParser | ||||
from .common import ( | from .common import ( | ||||
ACCEPT_ARCHIVE_CONTENT_TYPES, | ACCEPT_ARCHIVE_CONTENT_TYPES, | ||||
Show All 15 Lines | class EditMediaAPI(APIPost, APIPut, APIDelete): | ||||
parser_classes = ( | parser_classes = ( | ||||
SWHFileUploadZipParser, | SWHFileUploadZipParser, | ||||
SWHFileUploadTarParser, | SWHFileUploadTarParser, | ||||
) | ) | ||||
def process_put( | def process_put( | ||||
self, req, headers: ParsedRequestHeaders, collection_name: str, deposit_id: int | self, req, headers: ParsedRequestHeaders, collection_name: str, deposit_id: int | ||||
) -> Dict[str, Any]: | ) -> None: | ||||
"""Replace existing content for the existing deposit. | """Replace existing content for the existing deposit. | ||||
source: http://swordapp.github.io/SWORDv2-Profile/SWORDProfile.html#protocoloperations_editingcontent_binary # noqa | source: http://swordapp.github.io/SWORDv2-Profile/SWORDProfile.html#protocoloperations_editingcontent_binary # noqa | ||||
Returns: | Returns: | ||||
204 No content | 204 No content | ||||
""" | """ | ||||
if req.content_type not in ACCEPT_ARCHIVE_CONTENT_TYPES: | if req.content_type not in ACCEPT_ARCHIVE_CONTENT_TYPES: | ||||
msg = "Packaging format supported is restricted to %s" % ( | msg = "Packaging format supported is restricted to %s" % ( | ||||
", ".join(ACCEPT_ARCHIVE_CONTENT_TYPES) | ", ".join(ACCEPT_ARCHIVE_CONTENT_TYPES) | ||||
) | ) | ||||
raise DepositError(BAD_REQUEST, msg) | raise DepositError(BAD_REQUEST, msg) | ||||
return self._binary_upload( | self._binary_upload( | ||||
req, headers, collection_name, deposit_id=deposit_id, replace_archives=True | req, headers, collection_name, deposit_id=deposit_id, replace_archives=True | ||||
) | ) | ||||
def process_post( | def process_post( | ||||
self, | self, | ||||
req, | req, | ||||
headers: ParsedRequestHeaders, | headers: ParsedRequestHeaders, | ||||
collection_name: str, | collection_name: str, | ||||
Show All 17 Lines | ) -> Tuple[int, str, Dict]: | ||||
raise DepositError(BAD_REQUEST, msg) | raise DepositError(BAD_REQUEST, msg) | ||||
return ( | return ( | ||||
status.HTTP_201_CREATED, | status.HTTP_201_CREATED, | ||||
CONT_FILE_IRI, | CONT_FILE_IRI, | ||||
self._binary_upload(req, headers, collection_name, deposit_id), | self._binary_upload(req, headers, collection_name, deposit_id), | ||||
) | ) | ||||
def process_delete(self, req, collection_name: str, deposit_id: int) -> Dict: | def process_delete(self, req, collection_name: str, deposit_id: int) -> None: | ||||
"""Delete content (archives) from existing deposit. | """Delete content (archives) from existing deposit. | ||||
source: http://swordapp.github.io/SWORDv2-Profile/SWORDProfile.html#protocoloperations_deletingcontent # noqa | source: http://swordapp.github.io/SWORDv2-Profile/SWORDProfile.html#protocoloperations_deletingcontent # noqa | ||||
Returns: | Returns: | ||||
204 Created | 204 Created | ||||
""" | """ | ||||
return self._delete_archives(collection_name, deposit_id) | self._delete_archives(collection_name, deposit_id) |