diff --git a/tom_dataproducts/api_views.py b/tom_dataproducts/api_views.py index d875cb402..c239c7989 100644 --- a/tom_dataproducts/api_views.py +++ b/tom_dataproducts/api_views.py @@ -1,9 +1,15 @@ +import logging +from copy import deepcopy + from django.conf import settings +from django.core.exceptions import ValidationError as DjangoValidationError from django.http import Http404 from django_filters import rest_framework as drf_filters +from django.db import IntegrityError from guardian.mixins import PermissionListMixin from guardian.shortcuts import assign_perm, get_objects_for_user from rest_framework import status +from rest_framework.exceptions import ValidationError as DRFValidationError from rest_framework.mixins import CreateModelMixin, DestroyModelMixin, ListModelMixin from rest_framework.parsers import MultiPartParser, FormParser, JSONParser from rest_framework.response import Response @@ -14,11 +20,13 @@ from tom_dataproducts.filters import DataProductFilter, ReducedDatumFilter from tom_dataproducts.models import (DataProduct, ReducedDatum, PhotometryReducedDatum, SpectroscopyReducedDatum, AstrometryReducedDatum, - REDUCED_DATUM_MODELS) + REDUCED_DATUM_MODELS, try_parse_reduced_datum) from tom_dataproducts.serializers import DataProductSerializer, ReducedDatumSerializer from tom_targets.models import Target +logger = logging.getLogger(__name__) + # Maps the data_type query param to the concrete model that holds those rows. _DATA_TYPE_MODEL_MAP = { 'photometry': PhotometryReducedDatum, @@ -47,27 +55,55 @@ class DataProductViewSet(CreateModelMixin, DestroyModelMixin, ListModelMixin, Ge def create(self, request, *args, **kwargs): request.data['data'] = request.FILES['file'] - response = super().create(request, *args, **kwargs) - - if response.status_code == status.HTTP_201_CREATED: - response.data['message'] = 'Data product successfully uploaded.' - dp = DataProduct.objects.get(pk=response.data['id']) - try: - run_hook('data_product_post_upload', dp) - reduced_data = run_data_processor(dp) - if not settings.TARGET_PERMISSIONS_ONLY: - for group in response.data['group']: - assign_perm('tom_dataproducts.view_dataproduct', group, dp) - assign_perm('tom_dataproducts.delete_dataproduct', group, dp) - assign_perm('tom_dataproducts.view_reduceddatum', group, reduced_data) - except Exception: - for model in REDUCED_DATUM_MODELS: - model.objects.filter(data_product=dp).delete() - dp.delete() - return Response({'Data processing error': '''There was an error in processing your DataProduct into \ - individual ReducedDatum objects.'''}, - status=status.HTTP_500_INTERNAL_SERVER_ERROR) - return response + product_id = request.data.get('product_id') + + existing_dp = DataProduct.objects.filter(product_id=product_id).first() if product_id else None + if existing_dp: + logger.info('Skipping duplicate DataProduct upload for product_id=%s', product_id) + payload = self.get_serializer(existing_dp).data + payload['message'] = 'Data product already exists. Skipping upload.' + payload['already_exists'] = True + return Response(payload, status=status.HTTP_200_OK) + + serializer = self.get_serializer(data=request.data) + try: + serializer.is_valid(raise_exception=True) + except DRFValidationError as exc: + product_id_errors = exc.detail.get('product_id', []) if hasattr(exc, 'detail') else [] + has_duplicate_product_id_error = any('already exists' in str(err).lower() for err in product_id_errors) + if product_id and has_duplicate_product_id_error: + existing_dp = DataProduct.objects.filter(product_id=product_id).first() + if existing_dp: + logger.info('Skipping duplicate DataProduct upload during validation for product_id=%s', product_id) + payload = self.get_serializer(existing_dp).data + payload['message'] = 'Data product already exists. Skipping upload.' + payload['already_exists'] = True + return Response(payload, status=status.HTTP_200_OK) + raise + + self.perform_create(serializer) + payload = serializer.data + payload['message'] = 'Data product successfully uploaded.' + headers = self.get_success_headers(payload) + dp = DataProduct.objects.get(pk=payload['id']) + + try: + run_hook('data_product_post_upload', dp) + reduced_data = run_data_processor(dp) + if not settings.TARGET_PERMISSIONS_ONLY: + for group in payload['group']: + assign_perm('tom_dataproducts.view_dataproduct', group, dp) + assign_perm('tom_dataproducts.delete_dataproduct', group, dp) + assign_perm('tom_dataproducts.view_reduceddatum', group, reduced_data) + except Exception: + for model in REDUCED_DATUM_MODELS: + model.objects.filter(data_product=dp).delete() + dp.delete() + return Response({'Data processing error': '''There was an error in processing your DataProduct into \ + individual ReducedDatum objects.'''}, + status=status.HTTP_500_INTERNAL_SERVER_ERROR) + + return Response(payload, status=status.HTTP_201_CREATED, headers=headers) def get_queryset(self): """ @@ -152,10 +188,60 @@ def list(self, request, *args, **kwargs): return Response(self.get_serializer(all_instances, many=True).data) - def create(self, request, *args, **kwargs): - response = super().create(request, *args, **kwargs) - - if response.status_code == status.HTTP_201_CREATED: - response.data['message'] = 'Data successfully uploaded.' + def _find_existing_reduced_datum(self, validated_data): + """Return an existing duplicate ReducedDatum instance if one can be identified.""" + candidate = try_parse_reduced_datum(deepcopy(validated_data)) + + if isinstance(candidate, PhotometryReducedDatum): + return PhotometryReducedDatum.objects.filter( + target=candidate.target, + bandpass=candidate.bandpass, + timestamp=candidate.timestamp, + ).first() + if isinstance(candidate, SpectroscopyReducedDatum): + return SpectroscopyReducedDatum.objects.filter( + target=candidate.target, + timestamp=candidate.timestamp, + telescope=candidate.telescope, + instrument=candidate.instrument, + ).first() + if isinstance(candidate, AstrometryReducedDatum): + return AstrometryReducedDatum.objects.filter( + target=candidate.target, + timestamp=candidate.timestamp, + telescope=candidate.telescope, + instrument=candidate.instrument, + ).first() + + return ReducedDatum.objects.filter( + target=candidate.target, + data_type=candidate.data_type, + timestamp=candidate.timestamp, + value=candidate.value, + ).first() - return response + def create(self, request, *args, **kwargs): + serializer = self.get_serializer(data=request.data) + serializer.is_valid(raise_exception=True) + + try: + self.perform_create(serializer) + payload = serializer.data + payload['message'] = 'Data successfully uploaded.' + headers = self.get_success_headers(payload) + return Response(payload, status=status.HTTP_201_CREATED, headers=headers) + except (DjangoValidationError, IntegrityError): + existing = self._find_existing_reduced_datum(serializer.validated_data) + if existing is None: + raise + + logger.info( + 'Skipping duplicate ReducedDatum upload for target_id=%s timestamp=%s source_name=%s', + getattr(existing, 'target_id', None), + getattr(existing, 'timestamp', None), + getattr(existing, 'source_name', ''), + ) + payload = self.get_serializer(existing).data + payload['message'] = 'Data already exists. Skipping upload.' + payload['already_exists'] = True + return Response(payload, status=status.HTTP_200_OK) diff --git a/tom_dataproducts/tests/test_api.py b/tom_dataproducts/tests/test_api.py index 52a22e29e..638dea0e6 100644 --- a/tom_dataproducts/tests/test_api.py +++ b/tom_dataproducts/tests/test_api.py @@ -1,6 +1,6 @@ from tom_targets.base_models import get_target_model_app_label from django.contrib.auth.models import Group, User -from django.core.exceptions import ValidationError + from django.core.files.uploadedfile import SimpleUploadedFile from django.urls import reverse from guardian.shortcuts import assign_perm @@ -77,6 +77,25 @@ def test_data_product_upload_invalid_type(self): self.assertContains(response, 'Not a valid data_product_type.', status_code=status.HTTP_400_BAD_REQUEST) + def test_data_product_upload_duplicate_product_id_is_skipped(self): + with open('tom_dataproducts/tests/test_data/test_lightcurve.csv', 'rb') as first_lightcurve_file: + self.dp_data['file'] = first_lightcurve_file + first_response = self.client.post(reverse('api:dataproducts-list'), self.dp_data, format='multipart') + + self.assertEqual(first_response.status_code, status.HTTP_201_CREATED) + self.assertEqual(DataProduct.objects.count(), 1) + self.assertEqual(PhotometryReducedDatum.objects.count(), 3) + + with open('tom_dataproducts/tests/test_data/test_lightcurve.csv', 'rb') as second_lightcurve_file: + self.dp_data['file'] = second_lightcurve_file + second_response = self.client.post(reverse('api:dataproducts-list'), self.dp_data, format='multipart') + + self.assertEqual(second_response.status_code, status.HTTP_200_OK) + self.assertEqual(DataProduct.objects.count(), 1) + self.assertEqual(PhotometryReducedDatum.objects.count(), 3) + self.assertEqual(second_response.data['id'], first_response.data['id']) + self.assertTrue(second_response.data['already_exists']) + def test_data_product_upload_failed_processing(self): self.dp_data['data_product_type'] = 'spectroscopy' @@ -139,13 +158,19 @@ def test_upload_reduced_datum(self): def test_upload_same_reduced_datum_twice(self): """ - Test that identical data raises a validation error while similar but different JSON will make it through. + Test that identical data is skipped while similar but different JSON will make it through. """ - self.client.post(reverse('api:reduceddatums-list'), self.rd_data, format='json') - with self.assertRaises(ValidationError): - self.client.post(reverse('api:reduceddatums-list'), self.rd_data, format='json') + first_response = self.client.post(reverse('api:reduceddatums-list'), self.rd_data, format='json') + self.assertEqual(first_response.status_code, status.HTTP_201_CREATED) + + second_response = self.client.post(reverse('api:reduceddatums-list'), self.rd_data, format='json') + self.assertEqual(second_response.status_code, status.HTTP_200_OK) + self.assertTrue(second_response.data['already_exists']) + self.rd_data['value'] = {'magnitude': 15.582, 'filter': 'B', 'error': 0.005} - self.client.post(reverse('api:reduceddatums-list'), self.rd_data, format='json') + third_response = self.client.post(reverse('api:reduceddatums-list'), self.rd_data, format='json') + self.assertEqual(third_response.status_code, status.HTTP_201_CREATED) + rd_queryset = PhotometryReducedDatum.objects.all() self.assertEqual(rd_queryset.count(), 2)