refactor export data progress

Author: Andrei Neagu

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

@@ -17,7 +17,6 @@
 from ...modules.celery.models import TaskID, TaskId
 from ...modules.celery.utils import get_celery_worker, get_fastapi_app
 from ...simcore_s3_dsm import SimcoreS3DataManager
-from ...utils.progress_utils import get_tqdm_progress, set_tqdm_absolute_progress
 
 _logger = logging.getLogger(__name__)
 
@@ -69,34 +68,35 @@ async def export_data(
     user_id: UserID,
     paths_to_export: list[S3ObjectKey],
 ) -> StorageFileID:
-    _logger.info("Exporting (for user='%s') selection: %s", user_id, paths_to_export)
-
-    dsm = get_dsm_provider(get_fastapi_app(task.app)).get(
-        SimcoreS3DataManager.get_location_id()
-    )
-    assert isinstance(dsm, SimcoreS3DataManager)  # nosec
-
-    try:
-        for path_to_export in paths_to_export:
-            await dsm.can_read_file(user_id=user_id, file_id=path_to_export)
-    except FileAccessRightError as err:
-        raise AccessRightError(
-            user_id=user_id,
-            file_id=path_to_export,
-            location_id=SimcoreS3DataManager.get_location_id(),
-        ) from err
+    with log_context(
+        _logger,
+        logging.INFO,
+        f"'{task_id}' export data (for {user_id=}) fom selection: {paths_to_export}",
+    ):
+        dsm = get_dsm_provider(get_fastapi_app(task.app)).get(
+            SimcoreS3DataManager.get_location_id()
+        )
+        assert isinstance(dsm, SimcoreS3DataManager)  # nosec
 
-    with get_tqdm_progress(total=1, description=f"{task.name}") as pbar:
+        try:
+            for path_to_export in paths_to_export:
+                await dsm.can_read_file(user_id=user_id, file_id=path_to_export)
+        except FileAccessRightError as err:
+            raise AccessRightError(
+                user_id=user_id,
+                file_id=path_to_export,
+                location_id=SimcoreS3DataManager.get_location_id(),
+            ) from err
 
         async def _progress_cb(report: ProgressReport) -> None:
-            set_tqdm_absolute_progress(pbar, report)
             assert task.name  # nosec
-            await get_celery_worker(task.app).set_task_progress(
-                task.name, task_id, report
-            )
+            get_celery_worker(task.app).set_task_progress(task.name, task_id, report)
+            _logger.debug("'%s' progress %s", task_id, report.percent_value)
 
         async with ProgressBarData(
-            num_steps=1, description="data export", progress_report_cb=_progress_cb
+            num_steps=1,
+            description=f"'{task_id}' export data",
+            progress_report_cb=_progress_cb,
         ) as progress_bar:
             return await dsm.create_s3_export(
                 user_id, paths_to_export, progress_bar=progress_bar

services/storage/src/simcore_service_storage/modules/celery/worker.py

@@ -1,7 +1,6 @@
 import logging
 
 from celery import Celery  # type: ignore[import-untyped]
-from common_library.async_tools import make_async
 from models_library.progress_bar import ProgressReport
 from servicelib.logging_utils import log_context
 
@@ -14,7 +13,6 @@ class CeleryTaskQueueWorker:
     def __init__(self, celery_app: Celery) -> None:
         self.celery_app = celery_app
 
-    @make_async()
     def set_task_progress(
         self, task_name: str, task_id: TaskID, report: ProgressReport
     ) -> None:

services/storage/src/simcore_service_storage/utils/progress_utils.py

@@ -514,7 +514,7 @@ async def _request_start_data_export(
     rpc_client: RabbitMQRPCClient,
     user_id: UserID,
     product_name: ProductName,
-    paths_to_export: list[SimcoreS3FileID],
+    paths_to_export: list[Path],
     *,
     client_timeout: datetime.timedelta = datetime.timedelta(seconds=60),
 ) -> dict[str, Any]:
@@ -601,7 +601,7 @@ async def test_start_data_export(
         storage_rabbitmq_rpc_client,
         user_id,
         product_name,
-        paths_to_export=list(paths_to_export),
+        paths_to_export=[Path(x) for x in paths_to_export],
     )
 
     assert re.fullmatch(
@@ -628,7 +628,9 @@ async def test_start_data_export_access_error(
             storage_rabbitmq_rpc_client,
             user_id,
             product_name,
-            paths_to_export=[f"{faker.uuid4()}/{faker.uuid4()}/{faker.file_name()}"],
+            paths_to_export=[
+                Path(f"{faker.uuid4()}/{faker.uuid4()}/{faker.file_name()}")
+            ],
             client_timeout=datetime.timedelta(seconds=60),
         )