change task key prefix

Author: giancarloromeo

packages/celery-library/src/celery_library/backends/redis.py

@@ -21,7 +21,7 @@
 )
 from servicelib.redis import RedisClientSDK, handle_redis_returns_union_types
 
-_CELERY_TASK_INFO_PREFIX: Final[str] = "celery-task-info-"
+_CELERY_TASK_PREFIX: Final[str] = "celery-task-"
 _CELERY_TASK_STREAM_PREFIX: Final[str] = "celery-task-stream-"
 _CELERY_TASK_ID_KEY_ENCODING = "utf-8"
 _CELERY_TASK_SCAN_COUNT_PER_BATCH: Final[int] = 1000
@@ -38,16 +38,16 @@
 _logger = logging.getLogger(__name__)
 
 
-def _build_info_key(task_id: TaskID) -> str:
-    return _CELERY_TASK_INFO_PREFIX + task_id
+def _build_key(task_id: TaskID) -> str:
+    return _CELERY_TASK_PREFIX + task_id
 
 
 def _build_stream_key(task_id: TaskID) -> str:
     return _CELERY_TASK_STREAM_PREFIX + task_id
 
 
 @dataclass
-class RedisTaskInfoStore:
+class RedisTaskStore:
     _redis_client_sdk: RedisClientSDK
 
     async def create_task(
@@ -56,7 +56,7 @@ async def create_task(
         execution_metadata: ExecutionMetadata,
         expiry: timedelta,
     ) -> None:
-        task_key = _build_info_key(task_id)
+        task_key = _build_key(task_id)
         await handle_redis_returns_union_types(
             self._redis_client_sdk.redis.hset(
                 name=task_key,
@@ -88,7 +88,7 @@ async def create_task(
     async def get_task_metadata(self, task_id: TaskID) -> ExecutionMetadata | None:
         raw_result = await handle_redis_returns_union_types(
             self._redis_client_sdk.redis.hget(
-                _build_info_key(task_id), _CELERY_TASK_METADATA_KEY
+                _build_key(task_id), _CELERY_TASK_METADATA_KEY
             )
         )
         if not raw_result:
@@ -105,7 +105,7 @@ async def get_task_metadata(self, task_id: TaskID) -> ExecutionMetadata | None:
     async def get_task_progress(self, task_id: TaskID) -> ProgressReport | None:
         raw_result = await handle_redis_returns_union_types(
             self._redis_client_sdk.redis.hget(
-                _build_info_key(task_id), _CELERY_TASK_PROGRESS_KEY
+                _build_key(task_id), _CELERY_TASK_PROGRESS_KEY
             )
         )
         if not raw_result:
@@ -120,7 +120,7 @@ async def get_task_progress(self, task_id: TaskID) -> ProgressReport | None:
             return None
 
     async def list_tasks(self, owner_metadata: OwnerMetadata) -> list[Task]:
-        search_key = _CELERY_TASK_INFO_PREFIX + owner_metadata.model_dump_task_id(
+        search_key = _CELERY_TASK_PREFIX + owner_metadata.model_dump_task_id(
             task_uuid=WILDCARD
         )
 
@@ -157,20 +157,20 @@ async def list_tasks(self, owner_metadata: OwnerMetadata) -> list[Task]:
         return tasks
 
     async def remove_task(self, task_id: TaskID) -> None:
-        await self._redis_client_sdk.redis.delete(_build_info_key(task_id))
+        await self._redis_client_sdk.redis.delete(_build_key(task_id))
         await self._redis_client_sdk.redis.delete(_build_stream_key(task_id))
 
     async def set_task_progress(self, task_id: TaskID, report: ProgressReport) -> None:
         await handle_redis_returns_union_types(
             self._redis_client_sdk.redis.hset(
-                name=_build_info_key(task_id),
+                name=_build_key(task_id),
                 key=_CELERY_TASK_PROGRESS_KEY,
                 value=report.model_dump_json(),
             )
         )
 
     async def task_exists(self, task_id: TaskID) -> bool:
-        n = await self._redis_client_sdk.redis.exists(_build_info_key(task_id))
+        n = await self._redis_client_sdk.redis.exists(_build_key(task_id))
         assert isinstance(n, int)  # nosec
         return n > 0
 
@@ -209,4 +209,4 @@ async def consume_task_events(
 
 
 if TYPE_CHECKING:
-    _: type[TaskInfoStore] = RedisTaskInfoStore
+    _: type[TaskInfoStore] = RedisTaskStore

packages/celery-library/tests/conftest.py

@@ -15,7 +15,7 @@
 )
 from celery.signals import worker_init, worker_shutdown
 from celery.worker.worker import WorkController
-from celery_library.backends.redis import RedisTaskInfoStore
+from celery_library.backends.redis import RedisTaskStore
 from celery_library.signals import on_worker_init, on_worker_shutdown
 from celery_library.task_manager import CeleryTaskManager
 from celery_library.types import register_celery_types
@@ -66,7 +66,7 @@ async def run_until_shutdown(
         self._task_manager = CeleryTaskManager(
             self._app,
             self._settings,
-            RedisTaskInfoStore(redis_client_sdk),
+            RedisTaskStore(redis_client_sdk),
         )
 
         startup_completed_event.set()
@@ -173,7 +173,7 @@ async def celery_task_manager(
         yield CeleryTaskManager(
             mock_celery_app,
             celery_settings,
-            RedisTaskInfoStore(redis_client_sdk),
+            RedisTaskStore(redis_client_sdk),
         )
     finally:
         await redis_client_sdk.shutdown()

services/api-server/src/simcore_service_api_server/clients/celery_task_manager.py

@@ -1,6 +1,6 @@
 import logging
 
-from celery_library.backends.redis import RedisTaskInfoStore
+from celery_library.backends.redis import RedisTaskStore
 from celery_library.common import create_app
 from celery_library.task_manager import CeleryTaskManager
 from celery_library.types import register_celery_types, register_pydantic_types
@@ -30,7 +30,7 @@ async def on_startup() -> None:
             app.state.task_manager = CeleryTaskManager(
                 create_app(settings),
                 settings,
-                RedisTaskInfoStore(redis_client_sdk),
+                RedisTaskStore(redis_client_sdk),
             )
 
             register_celery_types()

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

@@ -1,6 +1,6 @@
 import logging
 
-from celery_library.backends.redis import RedisTaskInfoStore
+from celery_library.backends.redis import RedisTaskStore
 from celery_library.common import create_app
 from celery_library.task_manager import CeleryTaskManager
 from celery_library.types import register_celery_types, register_pydantic_types
@@ -34,7 +34,7 @@ async def on_startup() -> None:
             app.state.task_manager = CeleryTaskManager(
                 create_app(settings),
                 settings,
-                RedisTaskInfoStore(redis_client_sdk),
+                RedisTaskStore(redis_client_sdk),
             )
 
             register_celery_types()

services/web/server/src/simcore_service_webserver/celery/_task_manager.py

@@ -2,7 +2,7 @@
 from typing import Final
 
 from aiohttp import web
-from celery_library.backends.redis import RedisTaskInfoStore
+from celery_library.backends.redis import RedisTaskStore
 from celery_library.common import create_app
 from celery_library.task_manager import CeleryTaskManager
 from celery_library.types import register_celery_types
@@ -30,7 +30,7 @@ async def setup_task_manager(app: web.Application):
         app[_APP_CELERY_TASK_MANAGER_KEY] = CeleryTaskManager(
             celery_app,
             celery_settings,
-            RedisTaskInfoStore(redis_client_sdk),
+            RedisTaskStore(redis_client_sdk),
         )
         register_celery_types()