connected export job

Author: Andrei Neagu

services/storage/src/simcore_service_storage/api/_worker_tasks/_data_export.py

@@ -0,0 +1,40 @@
+import logging
+from typing import cast
+
+from celery import Task  # type: ignore[import-untyped]
+from models_library.progress_bar import ProgressReport
+from models_library.projects_nodes_io import StorageFileID
+from models_library.users import UserID
+
+from ...dsm import get_dsm_provider
+from ...modules.celery.utils import get_celery_worker, get_fastapi_app
+from ...simcore_s3_dsm import SimcoreS3DataManager
+from ._tqdm_utils import get_export_progress, set_absolute_progress
+
+_logger = logging.getLogger(__name__)
+
+
+async def data_export(
+    task: Task,
+    *,
+    user_id: UserID,
+    paths_to_export: list[StorageFileID],
+) -> StorageFileID:
+    _logger.info("Exporting for user '%s' files: %s", user_id, paths_to_export)
+
+    dsm = cast(
+        SimcoreS3DataManager,
+        get_dsm_provider(get_fastapi_app(task.app)).get(
+            SimcoreS3DataManager.get_location_id()
+        ),
+    )
+
+    with get_export_progress(total=1, description=f"{task.name}") as pbar:
+
+        async def _progress_cb(report: ProgressReport) -> None:
+            set_absolute_progress(pbar, current_progress=report.actual_value)
+            await get_celery_worker(task.app).set_task_progress(task, report)
+
+        return await dsm.create_s3_export(
+            user_id, paths_to_export, progress_cb=_progress_cb
+        )

services/storage/src/simcore_service_storage/api/_worker_tasks/_export_task.py

@@ -55,7 +55,8 @@ async def start_data_export(
     task_uuid = await get_celery_client(app).send_task(
         data_export.__name__,
         task_context=job_id_data.model_dump(),
-        files=data_export_start.file_and_folder_ids,
+        user_id=job_id_data.user_id,
+        paths_to_export=data_export_start.file_and_folder_ids,
     )
 
     return AsyncJobGet(