add Celery plugin

Author: giancarloromeo

services/web/server/src/simcore_service_webserver/application.py

@@ -21,6 +21,7 @@
 from .api_keys.plugin import setup_api_keys
 from .application_settings import get_application_settings, setup_settings
 from .catalog.plugin import setup_catalog
+from .celery.plugin import setup_celery
 from .collaboration.bootstrap import (
     setup_realtime_collaboration,
 )
@@ -188,6 +189,9 @@ def create_application() -> web.Application:
     setup_exporter(app)
     setup_realtime_collaboration(app)
 
+    # Celery
+    setup_celery(app)
+
     # NOTE: *last* events
     app.on_startup.append(_create_welcome_banner(WELCOME_MSG))
     app.on_shutdown.append(_create_finished_banner())

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

@@ -0,0 +1,3 @@
+from ._task_manager import get_task_manager
+
+__all__: tuple[str, ...] = ("get_task_manager",)

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

@@ -0,0 +1,41 @@
+import logging
+from typing import Final
+
+from aiohttp import web
+from celery_library.backends.redis import RedisTaskInfoStore
+from celery_library.common import create_app
+from celery_library.task_manager import CeleryTaskManager
+from celery_library.types import register_celery_types
+from servicelib.celery.task_manager import TaskManager
+from servicelib.logging_utils import log_context
+from settings_library.celery import CelerySettings
+
+from ..redis import get_redis_celery_tasks_client_sdk
+from .settings import get_plugin_settings
+
+_logger = logging.getLogger(__name__)
+
+_APP_CELERY_TASK_MANAGER_KEY: Final = web.AppKey(
+    CeleryTaskManager.__name__, CeleryTaskManager
+)
+
+
+async def setup_task_manager(app: web.Application):
+    with log_context(_logger, logging.INFO, "Setting up Celery task manager"):
+        celery_settings: CelerySettings = get_plugin_settings(app)
+
+        redis_client_sdk = get_redis_celery_tasks_client_sdk(app)
+        celery_app = create_app(celery_settings)
+
+        app[_APP_CELERY_TASK_MANAGER_KEY] = CeleryTaskManager(
+            celery_app,
+            celery_settings,
+            RedisTaskInfoStore(redis_client_sdk),
+        )
+        register_celery_types()
+
+    yield
+
+
+def get_task_manager(app: web.Application) -> TaskManager:
+    return app[_APP_CELERY_TASK_MANAGER_KEY]

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

@@ -0,0 +1,18 @@
+import logging
+
+from aiohttp import web
+
+from ..application_setup import ModuleCategory, app_setup_func
+from ._task_manager import setup_task_manager
+
+_logger = logging.getLogger(__name__)
+
+
+@app_setup_func(
+    __name__,
+    ModuleCategory.ADDON,
+    settings_name="WEBSERVER_CELERY",
+    logger=_logger,
+)
+def setup_celery(app: web.Application):
+    app.cleanup_ctx.append(setup_task_manager)

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

@@ -0,0 +1,11 @@
+from aiohttp import web
+from settings_library.celery import CelerySettings
+
+from ..application_keys import APP_SETTINGS_APPKEY
+
+
+def get_plugin_settings(app: web.Application) -> CelerySettings:
+    settings = app[APP_SETTINGS_APPKEY].WEBSERVER_CELERY
+    assert settings, "plugin.setup_celery not called?"  # nosec
+    assert isinstance(settings, CelerySettings)  # nosec
+    return settings