use Redis hash

Author: giancarloromeo

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

@@ -9,7 +9,7 @@
 from ...core.settings import get_application_settings
 from ._celery_types import register_celery_types
 from ._common import create_app
-from .backends._redis import RedisTaskMetadataStore
+from .backends._redis import RedisTaskInfoStore
 from .client import CeleryTaskQueueClient
 
 _logger = logging.getLogger(__name__)
@@ -31,7 +31,7 @@ async def on_startup() -> None:
         app.state.celery_client = CeleryTaskQueueClient(
             celery_app,
             celery_settings,
-            RedisTaskMetadataStore(redis_client_sdk),
+            RedisTaskInfoStore(redis_client_sdk),
         )
 
         register_celery_types()

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

@@ -7,19 +7,20 @@
 
 from ..models import TaskContext, TaskID, TaskMetadata, TaskUUID, build_task_id_prefix
 
-_CELERY_TASK_METADATA_PREFIX: Final[str] = "celery-task-metadata-"
+_CELERY_TASK_INFO_PREFIX: Final[str] = "celery-task-info-"
 _CELERY_TASK_ID_KEY_ENCODING = "utf-8"
 _CELERY_TASK_ID_KEY_SEPARATOR: Final[str] = ":"
 _CELERY_TASK_SCAN_COUNT_PER_BATCH: Final[int] = 10000
+_CELERY_TASK_METADATA_KEY: Final[str] = "metadata"
 
 _logger = logging.getLogger(__name__)
 
 
 def _build_key(task_id: TaskID) -> str:
-    return _CELERY_TASK_METADATA_PREFIX + task_id
+    return _CELERY_TASK_INFO_PREFIX + task_id
 
 
-class RedisTaskMetadataStore:
+class RedisTaskInfoStore:
     def __init__(self, redis_client_sdk: RedisClientSDK) -> None:
         self._redis_client_sdk = redis_client_sdk
 
@@ -28,13 +29,13 @@ async def exists(self, task_id: TaskID) -> bool:
         assert isinstance(n, int)  # nosec
         return n > 0
 
-    async def get(self, task_id: TaskID) -> TaskMetadata | None:
-        result = await self._redis_client_sdk.redis.get(_build_key(task_id))
+    async def get_metadata(self, task_id: TaskID) -> TaskMetadata | None:
+        result = await self._redis_client_sdk.redis.hget(_build_key(task_id), _CELERY_TASK_METADATA_KEY)  # type: ignore
         return TaskMetadata.model_validate_json(result) if result else None
 
     async def get_uuids(self, task_context: TaskContext) -> set[TaskUUID]:
         search_key = (
-            _CELERY_TASK_METADATA_PREFIX
+            _CELERY_TASK_INFO_PREFIX
             + build_task_id_prefix(task_context)
             + _CELERY_TASK_ID_KEY_SEPARATOR
         )
@@ -55,11 +56,15 @@ async def remove(self, task_id: TaskID) -> None:
         await self._redis_client_sdk.redis.delete(_build_key(task_id))
         AsyncResult(task_id).forget()
 
-    async def set(
+    async def set_metadata(
         self, task_id: TaskID, task_metadata: TaskMetadata, expiry: timedelta
     ) -> None:
-        await self._redis_client_sdk.redis.set(
+        await self._redis_client_sdk.redis.hset(
+            name=_build_key(task_id),
+            key=_CELERY_TASK_METADATA_KEY,
+            value=task_metadata.model_dump_json(),
+        )  # type: ignore
+        await self._redis_client_sdk.redis.expire(
             _build_key(task_id),
-            task_metadata.model_dump_json(),
-            ex=expiry,
+            expiry,
         )

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

@@ -17,8 +17,8 @@
 from .models import (
     TaskContext,
     TaskID,
+    TaskInfoStore,
     TaskMetadata,
-    TaskMetadataStore,
     TaskState,
     TaskStatus,
     TaskUUID,
@@ -46,7 +46,7 @@
 class CeleryTaskQueueClient:
     _celery_app: Celery
     _celery_settings: CelerySettings
-    _task_store: TaskMetadataStore
+    _task_store: TaskInfoStore
 
     async def send_task(
         self,
@@ -76,7 +76,7 @@ async def send_task(
                 if task_metadata.ephemeral
                 else self._celery_settings.CELERY_RESULT_EXPIRES
             )
-            await self._task_store.set(task_id, task_metadata, expiry=expiry)
+            await self._task_store.set_metadata(task_id, task_metadata, expiry=expiry)
             return task_uuid
 
     @make_async()
@@ -109,7 +109,7 @@ async def get_task_result(
             async_result = self._celery_app.AsyncResult(task_id)
             result = async_result.result
             if async_result.ready():
-                task_metadata = await self._task_store.get(task_id)
+                task_metadata = await self._task_store.get_metadata(task_id)
                 if task_metadata is not None and task_metadata.ephemeral:
                     await self._task_store.remove(task_id)
             return result

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

@@ -46,16 +46,16 @@ class TaskMetadata(BaseModel):
 _TASK_DONE = {TaskState.SUCCESS, TaskState.ERROR, TaskState.ABORTED}
 
 
-class TaskMetadataStore(Protocol):
+class TaskInfoStore(Protocol):
     async def exists(self, task_id: TaskID) -> bool: ...
 
-    async def get(self, task_id: TaskID) -> TaskMetadata | None: ...
+    async def get_metadata(self, task_id: TaskID) -> TaskMetadata | None: ...
 
     async def get_uuids(self, task_context: TaskContext) -> set[TaskUUID]: ...
 
     async def remove(self, task_id: TaskID) -> None: ...
 
-    async def set(
+    async def set_metadata(
         self, task_id: TaskID, task_data: TaskMetadata, expiry: timedelta
     ) -> None: ...