change boot process for celery worker

Author: bisgaard-itis

services/api-server/docker/boot.sh

@@ -48,15 +48,17 @@ if [ "${API_SERVER_WORKER_MODE}" = "true" ]; then
       --recursive \
       -- \
       celery \
-      --app=simcore_service_api_server.celery.worker_main:app \
+      --app=boot_celery_worker:app \
+      --workdir=services/api-server/docker \
       worker --pool=threads \
       --loglevel="${API_SERVER_LOGLEVEL}" \
       --concurrency="${CELERY_CONCURRENCY}" \
       --hostname="${API_SERVER_WORKER_NAME}" \
       --queues="${CELERY_QUEUES:-default}"
   else
     exec celery \
-      --app=simcore_service_api_server.celery.worker_main:app \
+      --app=boot_celery_worker:app \
+      --workdir=services/api-server/docker \
       worker --pool=threads \
       --loglevel="${API_SERVER_LOGLEVEL}" \
       --concurrency="${CELERY_CONCURRENCY}" \

services/api-server/docker/boot_celery_worker.py

@@ -0,0 +1,10 @@
+from celery.signals import worker_init, worker_shutdown  # type: ignore[import-untyped]
+from celery_library.signals import (
+    on_worker_shutdown,
+)
+from simcore_service_api_server.celery.worker_main import get_app, worker_init_wrapper
+
+app = get_app()
+
+worker_init.connect(worker_init_wrapper)
+worker_shutdown.connect(on_worker_shutdown)

services/api-server/src/simcore_service_api_server/celery/worker_main.py

@@ -2,11 +2,9 @@
 
 from functools import partial
 
-from celery.signals import worker_init, worker_shutdown  # type: ignore[import-untyped]
 from celery_library.common import create_app as create_celery_app
 from celery_library.signals import (
     on_worker_init,
-    on_worker_shutdown,
 )
 from servicelib.fastapi.celery.app_server import FastAPIAppServer
 from servicelib.logging_utils import setup_loggers
@@ -16,34 +14,29 @@
 from .worker_tasks.tasks import setup_worker_tasks
 
 
-def _get_settings() -> ApplicationSettings:
-    return ApplicationSettings.create_from_envs()
+def get_app():
+    _settings = ApplicationSettings.create_from_envs()
 
+    setup_loggers(
+        log_format_local_dev_enabled=_settings.API_SERVER_LOG_FORMAT_LOCAL_DEV_ENABLED,
+        logger_filter_mapping=_settings.API_SERVER_LOG_FILTER_MAPPING,
+        tracing_settings=_settings.API_SERVER_TRACING,
+        log_base_level=_settings.log_level,
+        noisy_loggers=None,
+    )
 
-_settings = _get_settings()
-
-setup_loggers(
-    log_format_local_dev_enabled=_settings.API_SERVER_LOG_FORMAT_LOCAL_DEV_ENABLED,
-    logger_filter_mapping=_settings.API_SERVER_LOG_FILTER_MAPPING,
-    tracing_settings=_settings.API_SERVER_TRACING,
-    log_base_level=_settings.log_level,
-    noisy_loggers=None,
-)
-
-assert _settings.API_SERVER_CELERY  # nosec
-app = create_celery_app(_settings.API_SERVER_CELERY)
+    assert _settings.API_SERVER_CELERY  # nosec
+    app = create_celery_app(_settings.API_SERVER_CELERY)
+    setup_worker_tasks(app)
 
-app_server = FastAPIAppServer(app=create_app(_settings))
+    return app
 
 
 def worker_init_wrapper(sender, **_kwargs):
+    _settings = ApplicationSettings.create_from_envs()
     assert _settings.API_SERVER_CELERY  # nosec
+    app_server = FastAPIAppServer(app=create_app(_settings))
+
     return partial(on_worker_init, app_server, _settings.API_SERVER_CELERY)(
         sender, **_kwargs
     )
-
-
-worker_init.connect(worker_init_wrapper)
-worker_shutdown.connect(on_worker_shutdown)
-
-setup_worker_tasks(app)