Ensure that the imported files will be uploaded after ingesting the data to the database

backend/data_import/celery_tasks.py

@@ -43,11 +43,10 @@ def import_dataset(user_id, project_id, file_format: str, upload_ids: List[str],
     cleaner = create_cleaner(project)
     writer = BulkWriter(batch_size=settings.IMPORT_BATCH_SIZE, save_names=save_names)
     writer.save(reader, project, user, cleaner)
+    upload_to_store(temporary_uploads)
     return {"error": writer.errors + errors}
 
 
-@shared_task
-def upload_to_store(upload_ids: List[int]):
-    temporary_uploads = TemporaryUpload.objects.filter(upload_id__in=upload_ids)
+def upload_to_store(temporary_uploads):
     for tu in temporary_uploads:
         store_upload(tu.upload_id, destination_file_path=tu.file.name)

backend/data_import/views.py

@@ -4,7 +4,7 @@ from rest_framework.permissions import IsAuthenticated
 from rest_framework.response import Response
 from rest_framework.views import APIView
 
-from .celery_tasks import import_dataset, upload_to_store
+from .celery_tasks import import_dataset
 from .pipeline.catalog import Options
 from projects.models import Project
 from projects.permissions import IsProjectAdmin
@@ -24,16 +24,13 @@ class DatasetImportAPI(APIView):
     permission_classes = [IsAuthenticated & IsProjectAdmin]
 
     def post(self, request, *args, **kwargs):
-        project_id = self.kwargs["project_id"]
         upload_ids = request.data.pop("uploadIds")
         file_format = request.data.pop("format")
-
         task = import_dataset.delay(
             user_id=request.user.id,
-            project_id=project_id,
+            project_id=self>.kwargs["project_id"],
             file_format=file_format,
             upload_ids=upload_ids,
             **request.data,
         )
-        upload_task = upload_to_store.delay(upload_ids)
-        return Response({"task_id": task.task_id, "uploadTaskId": upload_task.task_id})
+        return Response({"task_id": task.task_id})