refactor

Author: giancarloromeo

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

@@ -1,5 +1,3 @@
-from uuid import uuid4
-
 from fastapi import FastAPI
 from models_library.api_schemas_rpc_async_jobs.async_jobs import AsyncJobGet, AsyncJobId
 from models_library.api_schemas_storage.data_export_async_jobs import (
@@ -10,6 +8,9 @@
 )
 from servicelib.rabbitmq import RPCRouter
 
+from ...modules.celery.client import CeleryTaskQueueClient, TaskIDParts
+from ...modules.celery.utils import get_celery_client
+
 router = RPCRouter()
 
 
@@ -24,7 +25,18 @@ async def start_data_export(
     app: FastAPI, paths: DataExportTaskStartInput
 ) -> AsyncJobGet:
     assert app  # nosec
+
+    client: CeleryTaskQueueClient = get_celery_client(app)
+
+    task_id = await client.send_task(
+        task_name="sync_archive",
+        task_id_parts=TaskIDParts(
+            user_id=paths.user_id, product_name=paths.product_name
+        ),
+        files=paths.paths,
+    )
+
     return AsyncJobGet(
-        job_id=AsyncJobId(f"{uuid4()}"),
+        job_id=AsyncJobId(task_id),
         job_name=", ".join(str(p) for p in paths.paths),
     )

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

@@ -10,28 +10,22 @@
 
 from .models import TaskID, TaskIDParts, TaskStatus
 
-_PREFIX: Final = "ct"
-
 _logger = logging.getLogger(__name__)
 
-
-def _get_task_id_components(task_id_parts: TaskIDParts) -> list[str]:
-    return sorted(map(str, task_id_parts.values()))
-
-
-def _get_components_prefix(name: str, task_id_parts: TaskIDParts) -> list[str]:
-    return [_PREFIX, name, *_get_task_id_components(task_id_parts)]
+_CELERY_TASK_META_PREFIX = "celery-task-meta-"
+_PREFIX: Final[str] = "ct"
 
 
-def _get_task_id_prefix(name: str, task_id_parts: TaskIDParts) -> TaskID:
-    return "::".join(_get_components_prefix(name, task_id_parts))
+def _build_parts_prefix(name: str, task_id_parts: TaskIDParts) -> list[str]:
+    return [_PREFIX, name, *[f"{task_id_parts[key]}" for key in sorted(task_id_parts)]]
 
 
-def _get_task_id(name: str, task_id_parts: TaskIDParts) -> TaskID:
-    return "::".join([*_get_components_prefix(name, task_id_parts), f"{uuid4()}"])
+def build_task_id_prefix(name: str, task_id_parts: TaskIDParts) -> TaskID:
+    return "::".join(_build_parts_prefix(name, task_id_parts))
 
 
-_CELERY_TASK_META_PREFIX = "celery-task-meta-"
+def build_task_id(name: str, task_id_parts: TaskIDParts) -> TaskID:
+    return "::".join([*_build_parts_prefix(name, task_id_parts), f"{uuid4()}"])
 
 
 class CeleryTaskQueueClient:
@@ -42,7 +36,7 @@ def __init__(self, celery_app: Celery):
     def send_task(
         self, task_name: str, *, task_id_parts: TaskIDParts, **task_params
     ) -> TaskID:
-        task_id = _get_task_id(task_name, task_id_parts)
+        task_id = build_task_id(task_name, task_id_parts)
         _logger.debug("Submitting task %s: %s", task_name, task_id)
         task = self._celery_app.send_task(
             task_name, task_id=task_id, kwargs=task_params
@@ -84,7 +78,7 @@ def _get_completed_task_ids(
     ) -> list[TaskID]:
         search_key = (
             _CELERY_TASK_META_PREFIX
-            + _get_task_id_prefix(task_name, task_id_parts)
+            + build_task_id_prefix(task_name, task_id_parts)
             + "*"
         )
         redis = self._celery_app.backend.client