setup task worker

Author: giancarloromeo

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

@@ -58,7 +58,7 @@ async def get_uuids(self, task_context: TaskContext) -> set[TaskUUID]: ...
     async def remove(self, task_id: TaskID) -> None: ...
 
     async def set_metadata(
-        self, task_id: TaskID, task_data: TaskMetadata, expiry: timedelta
+        self, task_id: TaskID, task_metadata: TaskMetadata, expiry: timedelta
     ) -> None: ...
 
     async def set_progress(self, task_id: TaskID, report: ProgressReport) -> None: ...

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

@@ -8,10 +8,13 @@
 from celery import Celery  # type: ignore[import-untyped]
 from fastapi import FastAPI
 from servicelib.logging_utils import log_context
+from servicelib.redis._client import RedisClientSDK
+from settings_library.redis import RedisDatabase
+from simcore_service_storage._meta import APP_NAME
 
 from ...core.application import create_app
 from ...core.settings import ApplicationSettings
-from ...modules.celery import set_event_loop
+from ...modules.celery import RedisTaskInfoStore, set_event_loop
 from ...modules.celery.utils import (
     get_fastapi_app,
     set_celery_worker,
@@ -28,14 +31,30 @@
 def on_worker_init(sender, **_kwargs) -> None:
     startup_complete_event = threading.Event()
 
-    def _init_fastapi(startup_complete_event: threading.Event) -> None:
+    def _init(startup_complete_event: threading.Event) -> None:
         loop = asyncio.new_event_loop()
         asyncio.set_event_loop(loop)
         shutdown_event = asyncio.Event()
 
-        fastapi_app = create_app(ApplicationSettings.create_from_envs())
+        app_settings = ApplicationSettings.create_from_envs()
+        fastapi_app = create_app(app_settings)
 
-        async def lifespan(
+        assert app_settings.STORAGE_CELERY
+        celery_settings = app_settings.STORAGE_CELERY
+
+        async def setup_task_worker():
+            redis_client_sdk = RedisClientSDK(
+                celery_settings.CELERY_REDIS_RESULT_BACKEND.build_redis_dsn(
+                    RedisDatabase.CELERY_TASKS
+                ),
+                client_name=f"{APP_NAME}.celery_tasks",
+            )
+
+            set_celery_worker(
+                sender.app, CeleryTaskWorker(RedisTaskInfoStore(redis_client_sdk))
+            )
+
+        async def fastapi_lifespan(
             startup_complete_event: threading.Event, shutdown_event: asyncio.Event
         ) -> None:
             async with LifespanManager(
@@ -54,12 +73,14 @@ async def lifespan(
         set_event_loop(fastapi_app, loop)
 
         set_fastapi_app(sender.app, fastapi_app)
-        set_celery_worker(sender.app, CeleryTaskWorker(sender.app))
-        loop.run_until_complete(lifespan(startup_complete_event, shutdown_event))
+        loop.run_until_complete(setup_task_worker())
+        loop.run_until_complete(
+            fastapi_lifespan(startup_complete_event, shutdown_event)
+        )
 
     thread = threading.Thread(
         group=None,
-        target=_init_fastapi,
+        target=_init,
         name="fastapi_app",
         args=(startup_complete_event,),
         daemon=True,