fixed export task

Author: Andrei Neagu

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

@@ -5,7 +5,7 @@
 
 from ...modules.celery._celery_types import register_celery_types
 from ...modules.celery._task import define_task
-from ...modules.celery.tasks import export_data
+from ._data_export import data_export
 from ._paths import compute_path_size
 
 _logger = logging.getLogger(__name__)
@@ -18,5 +18,5 @@ def setup_worker_tasks(app: Celery) -> None:
         logging.INFO,
         msg="Storage setup Worker Tasks",
     ):
-        define_task(app, export_data)
+        define_task(app, data_export)
         define_task(app, compute_path_size)

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

@@ -17,6 +17,7 @@
 from ...modules.celery import get_celery_client
 from ...modules.datcore_adapter.datcore_adapter_exceptions import DatcoreAdapterError
 from ...simcore_s3_dsm import SimcoreS3DataManager
+from .._worker_tasks._data_export import data_export
 
 router = RPCRouter()
 
@@ -52,9 +53,9 @@ async def start_data_export(
         ) from err
 
     task_uuid = await get_celery_client(app).send_task(
-        "export_data",
+        data_export.__name__,
         task_context=job_id_data.model_dump(),
-        files=data_export_start.file_and_folder_ids,  # ANE: adapt here your signature
+        files=data_export_start.file_and_folder_ids,
     )
 
     return AsyncJobGet(

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

@@ -1,28 +1,34 @@
 import logging
 
-from celery import Celery  # type: ignore[import-untyped]
+from celery import Celery, Task  # 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
 
-from .models import TaskID
-
 _logger = logging.getLogger(__name__)
 
 
 class CeleryTaskQueueWorker:
     def __init__(self, celery_app: Celery) -> None:
         self.celery_app = celery_app
 
-    def set_task_progress(
-        self, task_name: str, task_id: TaskID, report: ProgressReport
-    ) -> None:
+    @make_async()
+    def set_task_progress(self, task: Task, report: ProgressReport) -> None:
+        assert task.name  # nosec
+        task_name = task.name
+        task_id = task.request.id
+
         with log_context(
             _logger,
             logging.DEBUG,
             msg=f"Setting progress for {task_name}: {report.model_dump_json()}",
         ):
-            self.celery_app.tasks[task_name].update_state(
-                task_id=task_id,
-                state="RUNNING",
-                meta=report.model_dump(mode="json"),
-            )
+            try:
+                self.celery_app.tasks[task_name].update_state(
+                    task_id=task_id,
+                    state="RUNNING",
+                    meta=report.model_dump(mode="json"),
+                )
+            except ValueError as e:
+                if "task_id must not be empty. Got None instead." not in f"{e}":
+                    raise