adapt code

Author: giancarloromeo

packages/models-library/src/models_library/api_schemas_rpc_async_jobs/async_jobs.py

@@ -13,7 +13,7 @@
 
 class AsyncJobStatus(BaseModel):
     job_id: AsyncJobId
-    progress: ProgressReport
+    progress: ProgressReport | None
     done: bool
     started: datetime
     stopped: datetime | None

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

@@ -1,6 +1,5 @@
 # pylint: disable=unused-argument
 from datetime import datetime
-from uuid import uuid4
 
 from fastapi import FastAPI
 from models_library.api_schemas_rpc_async_jobs.async_jobs import (
@@ -15,9 +14,11 @@
     ResultError,
     StatusError,
 )
-from models_library.progress_bar import ProgressReport
 from servicelib.rabbitmq import RPCRouter
 
+from ...modules.celery.models import TaskStatus, TaskUUID
+from ...modules.celery.utils import get_celery_client
+
 router = RPCRouter()
 
 
@@ -36,13 +37,17 @@ async def get_status(
 ) -> AsyncJobStatus:
     assert app  # nosec
     assert job_id_data  # nosec
-    progress_report = ProgressReport(actual_value=0.5, total=1.0, attempt=1)
+
+    task_status: TaskStatus = await get_celery_client(app).get_task_status(
+        task_context=job_id_data.model_dump(),
+        task_uuid=job_id,
+    )
     return AsyncJobStatus(
         job_id=job_id,
-        progress=progress_report,
+        progress=task_status.progress_report,
         done=False,
-        started=datetime.now(),
-        stopped=None,
+        started=datetime.now(),  # TODO: retrieve that
+        stopped=None,  # TODO: retrieve that
     )
 
 
@@ -53,12 +58,23 @@ async def get_result(
     assert app  # nosec
     assert job_id  # nosec
     assert job_id_data  # nosec
-    return AsyncJobResult(result="Here's your result.", error=None)
+
+    result = await get_celery_client(app).get_result(
+        task_context=job_id_data.model_dump(),
+        task_uuid=job_id,
+    )
+
+    return AsyncJobResult(result=result, error=None)
 
 
 @router.expose()
 async def list_jobs(
-    app: FastAPI, filter_: str, job_id_data: AsyncJobNameData
+    app: FastAPI, filter_: str, job_id_data: AsyncJobNameData  # TODO: implement filter
 ) -> list[AsyncJobGet]:
     assert app  # nosec
-    return [AsyncJobGet(job_id=AsyncJobId(f"{uuid4()}"))]
+
+    task_uuids: set[TaskUUID] = await get_celery_client(app).get_task_uuids(
+        task_context=job_id_data.model_dump(),
+    )
+
+    return [AsyncJobGet(job_id=task_uuid) for task_uuid in task_uuids]

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

@@ -1,9 +1,6 @@
-from uuid import uuid4
-
 from fastapi import FastAPI
 from models_library.api_schemas_rpc_async_jobs.async_jobs import (
     AsyncJobGet,
-    AsyncJobId,
     AsyncJobNameData,
 )
 from models_library.api_schemas_storage.data_export_async_jobs import (
@@ -17,6 +14,7 @@
 from ...datcore_dsm import DatCoreDataManager
 from ...dsm import get_dsm_provider
 from ...exceptions.errors import FileAccessRightError
+from ...modules.celery.utils import get_celery_client
 from ...modules.datcore_adapter.datcore_adapter_exceptions import DatcoreAdapterError
 from ...simcore_s3_dsm import SimcoreS3DataManager
 
@@ -53,6 +51,10 @@ async def start_data_export(
             location_id=data_export_start.location_id,
         ) from err
 
+    task_uuid = await get_celery_client(app).send_task(
+        "export_data", task_context=job_id_data.model_dump()
+    )
+
     return AsyncJobGet(
-        job_id=AsyncJobId(f"{uuid4()}"),
+        job_id=task_uuid,
     )

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

@@ -105,7 +105,7 @@ def _get_completed_task_ids(self, task_context: TaskContext) -> set[TaskUUID]:
         return set()
 
     @make_async()
-    def get_task_ids(self, task_context: TaskContext) -> set[TaskUUID]:
+    def get_task_uuids(self, task_context: TaskContext) -> set[TaskUUID]:
         all_task_ids = self._get_completed_task_ids(task_context)
 
         for task_inspect_status in _CELERY_INSPECT_TASK_STATUSES: