fix methods

Author: giancarloromeo

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

@@ -25,17 +25,18 @@ async def compute_path_size(
     location_id: LocationID,
     path: Path,
 ) -> AsyncJobGet:
+    task_name = remote_compute_path_size.__name__
     task_uuid = await get_celery_client(app).submit_task(
         task_metadata=TaskMetadata(
-            name=remote_compute_path_size.__name__,
+            name=task_name,
         ),
         task_context=job_id_data.model_dump(),
         user_id=job_id_data.user_id,
         location_id=location_id,
         path=path,
     )
 
-    return AsyncJobGet(job_id=task_uuid)
+    return AsyncJobGet(job_id=task_uuid, job_name=task_name)
 
 
 @router.expose(reraise_if_error_type=None)
@@ -45,13 +46,14 @@ async def delete_paths(
     location_id: LocationID,
     paths: set[Path],
 ) -> AsyncJobGet:
+    task_name = remote_delete_paths.__name__
     task_uuid = await get_celery_client(app).submit_task(
         task_metadata=TaskMetadata(
-            name=remote_delete_paths.__name__,
+            name=task_name,
         ),
         task_context=job_id_data.model_dump(),
         user_id=job_id_data.user_id,
         location_id=location_id,
         paths=paths,
     )
-    return AsyncJobGet(job_id=task_uuid)
+    return AsyncJobGet(job_id=task_uuid, job_name=task_name)

services/storage/src/simcore_service_storage/modules/celery/backends/_redis.py

@@ -69,8 +69,9 @@ async def list_tasks(self, task_context: TaskContext) -> list[Task]:
             + build_task_id_prefix(task_context)
             + _CELERY_TASK_ID_KEY_SEPARATOR
         )
+        search_key_len = len(search_key)
+
         keys: list[str] = []
-        tasks = []
         pipe = self._redis_client_sdk.redis.pipeline()
         async for key in self._redis_client_sdk.redis.scan_iter(
             match=search_key + "*", count=_CELERY_TASK_SCAN_COUNT_PER_BATCH
@@ -82,19 +83,18 @@ async def list_tasks(self, task_context: TaskContext) -> list[Task]:
                 else key
             )
             keys.append(_key)
-
-        for key in keys:
-            pipe.hget(key, _CELERY_TASK_METADATA_KEY)
+            pipe.hget(_key, _CELERY_TASK_METADATA_KEY)
 
         results = await pipe.execute()
-        for key, task_metadata in zip(keys, results, strict=False):
-            tasks.append(
-                Task(
-                    uuid=TaskUUID(key.removeprefix(search_key)),
-                    metadata=TaskMetadata.model_validate_json(task_metadata),
-                )
+
+        return [
+            Task(
+                uuid=TaskUUID(key[search_key_len:]),
+                metadata=TaskMetadata.model_validate_json(metadata),
             )
-        return tasks
+            for key, metadata in zip(keys, results, strict=True)
+            if metadata is not None
+        ]
 
     async def remove_task(self, task_id: TaskID) -> None:
         await self._redis_client_sdk.redis.delete(_build_key(task_id))  # type: ignore

services/storage/tests/unit/test_modules_celery.py

@@ -23,6 +23,7 @@
 from simcore_service_storage.modules.celery.errors import TransferrableCeleryError
 from simcore_service_storage.modules.celery.models import (
     TaskContext,
+    TaskMetadata,
     TaskState,
 )
 from simcore_service_storage.modules.celery.utils import (
@@ -111,7 +112,9 @@ async def test_submitting_task_calling_async_function_results_with_success_state
     task_context = TaskContext(user_id=42)
 
     task_uuid = await celery_client.submit_task(
-        fake_file_processor.__name__,
+        TaskMetadata(
+            name=fake_file_processor.__name__,
+        ),
         task_context=task_context,
         files=[f"file{n}" for n in range(5)],
     )
@@ -139,7 +142,10 @@ async def test_submitting_task_with_failure_results_with_error(
     task_context = TaskContext(user_id=42)
 
     task_uuid = await celery_client.submit_task(
-        failure_task.__name__, task_context=task_context
+        TaskMetadata(
+            name=failure_task.__name__,
+        ),
+        task_context=task_context,
     )
 
     for attempt in Retrying(
@@ -162,7 +168,9 @@ async def test_aborting_task_results_with_aborted_state(
     task_context = TaskContext(user_id=42)
 
     task_uuid = await celery_client.submit_task(
-        dreamer_task.__name__,
+        TaskMetadata(
+            name=dreamer_task.__name__,
+        ),
         task_context=task_context,
     )
 
@@ -188,7 +196,9 @@ async def test_listing_task_uuids_contains_submitted_task(
     task_context = TaskContext(user_id=42)
 
     task_uuid = await celery_client.submit_task(
-        dreamer_task.__name__,
+        TaskMetadata(
+            name=dreamer_task.__name__,
+        ),
         task_context=task_context,
     )