add celery task

Author: giancarloromeo

services/storage/src/simcore_service_storage/core/application.py

@@ -29,6 +29,7 @@
 from ..dsm import setup_dsm
 from ..dsm_cleaner import setup_dsm_cleaner
 from ..exceptions.handlers import set_exception_handlers
+from ..modules.celery.celery import setup_celery
 from ..modules.db import setup_db
 from ..modules.long_running_tasks import setup_rest_api_long_running_tasks_for_uploads
 from ..modules.redis import setup as setup_redis
@@ -81,11 +82,14 @@ def create_app(settings: ApplicationSettings) -> FastAPI:
     setup_rest_api_routes(app, API_VTAG)
     set_exception_handlers(app)
 
+    setup_redis(app)
+
     setup_dsm(app)
     if settings.STORAGE_CLEANER_INTERVAL_S:
-        setup_redis(app)
         setup_dsm_cleaner(app)
 
+    setup_celery(app)
+
     if settings.STORAGE_PROFILING:
         app.add_middleware(ProfilerMiddleware)
 

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

@@ -0,0 +1,48 @@
+import logging
+from multiprocessing import Process
+from typing import cast
+
+from celery import Celery
+from celery.apps.worker import Worker
+from fastapi import FastAPI
+from settings_library.redis import RedisDatabase
+from simcore_service_storage.modules.celery.tasks import setup_celery_tasks
+
+from ...core.settings import get_application_settings
+
+_log = logging.getLogger(__name__)
+
+
+def setup_celery(app: FastAPI) -> None:
+    async def on_startup() -> None:
+        settings = get_application_settings(app)
+        assert settings.STORAGE_REDIS
+
+        redis_dsn = settings.STORAGE_REDIS.build_redis_dsn(
+            RedisDatabase.CELERY_TASKS,
+        )
+
+        app.state.celery_app = Celery(
+            broker=redis_dsn,
+            backend=redis_dsn,
+        )
+
+        setup_celery_tasks(app.state.celery_app)
+
+        # FIXME: Experiment: to start worker in a separate process
+        def worker_process():
+            worker = Worker(app=app.state.celery_app)
+            worker.start()
+
+        worker_proc = Process(target=worker_process)
+        worker_proc.start()
+
+    async def on_shutdown() -> None:
+        _log.warning("Implementing shutdown of celery app")
+
+    app.add_event_handler("startup", on_startup)
+    app.add_event_handler("shutdown", on_shutdown)
+
+
+def get_celery_app(app: FastAPI) -> Celery:
+    return cast(Celery, app.state.celery_app)

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

@@ -0,0 +1,13 @@
+import logging
+
+from celery import Celery
+
+_log = logging.getLogger(__name__)
+
+
+def archive(files: list[str]) -> None:
+    _log.info(f"Archiving {files=}")
+
+
+def setup_celery_tasks(celery_app: Celery) -> None:
+    celery_app.task()(archive)