Changeset View
Changeset View
Standalone View
Standalone View
swh/deposit/api/collection.py
| Show First 20 Lines • Show All 86 Lines • ▼ Show 20 Lines | ) -> Tuple[int, str, Receipt]: | ||||
| identifier | identifier | ||||
| - 400 (bad request) if the request's body is empty | - 400 (bad request) if the request's body is empty | ||||
| - 415 (unsupported media type) if a wrong media type is | - 415 (unsupported media type) if a wrong media type is | ||||
| provided | provided | ||||
| """ | """ | ||||
| assert deposit is None | assert deposit is None | ||||
| deposit = self._deposit_create(collection_name, external_id=headers.slug) | deposit = self._deposit_create(req, collection_name, external_id=headers.slug) | ||||
| if req.content_type in ACCEPT_ARCHIVE_CONTENT_TYPES: | if req.content_type in ACCEPT_ARCHIVE_CONTENT_TYPES: | ||||
| receipt = self._binary_upload( | receipt = self._binary_upload( | ||||
| req, headers, collection_name, deposit, check_slug_is_present=True | req, headers, collection_name, deposit, check_slug_is_present=True | ||||
| ) | ) | ||||
| elif req.content_type.startswith("multipart/"): | elif req.content_type.startswith("multipart/"): | ||||
| receipt = self._multipart_upload( | receipt = self._multipart_upload( | ||||
| req, headers, collection_name, deposit, check_slug_is_present=True | req, headers, collection_name, deposit, check_slug_is_present=True | ||||
| ) | ) | ||||
| else: | else: | ||||
| receipt = self._atom_entry( | receipt = self._atom_entry( | ||||
| req, headers, collection_name, deposit, check_slug_is_present=True | req, headers, collection_name, deposit, check_slug_is_present=True | ||||
| ) | ) | ||||
| return status.HTTP_201_CREATED, EDIT_IRI, receipt | return status.HTTP_201_CREATED, EDIT_IRI, receipt | ||||
| def _deposit_create( | def _deposit_create( | ||||
| self, collection_name: str, external_id: Optional[str] | self, request, collection_name: str, external_id: Optional[str] | ||||
| ) -> Deposit: | ) -> Deposit: | ||||
| collection = get_collection_by_name(collection_name) | collection = get_collection_by_name(collection_name) | ||||
| client = self.get_client(request) | |||||
| deposit_parent: Optional[Deposit] = None | deposit_parent: Optional[Deposit] = None | ||||
| assert client | |||||
| if external_id: | if external_id: | ||||
| try: | try: | ||||
| # find a deposit parent (same external id, status load to success) | # find a deposit parent (same external id, status load to success) | ||||
| deposit_parent = ( | deposit_parent = ( | ||||
| Deposit.objects.filter( | Deposit.objects.filter( | ||||
| client=self._client, | client=client, | ||||
| external_id=external_id, | external_id=external_id, | ||||
| status=DEPOSIT_STATUS_LOAD_SUCCESS, | status=DEPOSIT_STATUS_LOAD_SUCCESS, | ||||
| ) | ) | ||||
| .order_by("-id")[0:1] | .order_by("-id")[0:1] | ||||
| .get() | .get() | ||||
| ) | ) | ||||
| except Deposit.DoesNotExist: | except Deposit.DoesNotExist: | ||||
| # then no parent for that deposit, deposit_parent already None | # then no parent for that deposit, deposit_parent already None | ||||
| pass | pass | ||||
| return Deposit( | return Deposit( | ||||
| collection=collection, | collection=collection, | ||||
| external_id=external_id or "", | external_id=external_id or "", | ||||
| client=self._client, | client=client, | ||||
| parent=deposit_parent, | parent=deposit_parent, | ||||
| ) | ) | ||||