move method

Author: giancarloromeo

packages/celery-library/src/celery_library/__init__.py

@@ -2,12 +2,9 @@
 from asyncio import AbstractEventLoop
 
 from fastapi import FastAPI
-from servicelib.redis._client import RedisClientSDK
 from settings_library.celery import CelerySettings
-from settings_library.redis import RedisDatabase
 
-from .backends._redis import RedisTaskInfoStore
-from .common import create_app
+from .common import create_app, create_task_manager
 from .task_manager import CeleryTaskManager
 from .types import register_celery_types
 
@@ -16,18 +13,8 @@
 
 def setup_celery_client(app: FastAPI, celery_settings: CelerySettings) -> None:
     async def on_startup() -> None:
-        celery_app = create_app(celery_settings)
-        redis_client_sdk = RedisClientSDK(
-            celery_settings.CELERY_REDIS_RESULT_BACKEND.build_redis_dsn(
-                RedisDatabase.CELERY_TASKS
-            ),
-            client_name=f"{app.title}.celery_tasks",
-        )
-
-        app.state.celery_client = CeleryTaskManager(
-            celery_app,
-            celery_settings,
-            RedisTaskInfoStore(redis_client_sdk),
+        app.state.celery_client = create_task_manager(
+            create_app(celery_settings), celery_settings
         )
 
         register_celery_types()

packages/celery-library/src/celery_library/common.py

@@ -1,12 +1,13 @@
-import logging
 import ssl
 from typing import Any
 
 from celery import Celery  # type: ignore[import-untyped]
+from servicelib.redis import RedisClientSDK
 from settings_library.celery import CelerySettings
 from settings_library.redis import RedisDatabase
 
-_logger = logging.getLogger(__name__)
+from .backends._redis import RedisTaskInfoStore
+from .task_manager import CeleryTaskManager
 
 
 def _celery_configure(celery_settings: CelerySettings) -> dict[str, Any]:
@@ -25,13 +26,28 @@ def _celery_configure(celery_settings: CelerySettings) -> dict[str, Any]:
     return base_config
 
 
-def create_app(celery_settings: CelerySettings) -> Celery:
-    assert celery_settings
+def create_app(settings: CelerySettings) -> Celery:
+    assert settings
 
     return Celery(
-        broker=celery_settings.CELERY_RABBIT_BROKER.dsn,
-        backend=celery_settings.CELERY_REDIS_RESULT_BACKEND.build_redis_dsn(
+        broker=settings.CELERY_RABBIT_BROKER.dsn,
+        backend=settings.CELERY_REDIS_RESULT_BACKEND.build_redis_dsn(
             RedisDatabase.CELERY_TASKS,
         ),
-        **_celery_configure(celery_settings),
+        **_celery_configure(settings),
+    )
+
+
+def create_task_manager(app: Celery, settings: CelerySettings) -> CeleryTaskManager:
+    redis_client_sdk = RedisClientSDK(
+        settings.CELERY_REDIS_RESULT_BACKEND.build_redis_dsn(
+            RedisDatabase.CELERY_TASKS
+        ),
+        client_name="celery_tasks",
+    )
+
+    return CeleryTaskManager(
+        app,
+        settings,
+        RedisTaskInfoStore(redis_client_sdk),
     )

packages/celery-library/src/celery_library/signals.py

@@ -10,13 +10,10 @@
 from celery.worker.worker import WorkController  # type: ignore[import-untyped]
 from fastapi import FastAPI
 from servicelib.logging_utils import log_context
-from servicelib.redis._client import RedisClientSDK
 from settings_library.celery import CelerySettings
-from settings_library.redis import RedisDatabase
 
 from . import set_event_loop
-from .backends._redis import RedisTaskInfoStore
-from .task_manager import CeleryTaskManager
+from .common import create_task_manager
 from .utils import (
     get_fastapi_app,
     set_fastapi_app,
@@ -45,22 +42,14 @@ def _init(startup_complete_event: threading.Event) -> None:
         fastapi_app = app_factory()
         assert isinstance(fastapi_app, FastAPI)  # nosec
 
-        async def setup_task_worker():
-            redis_client_sdk = RedisClientSDK(
-                celery_settings.CELERY_REDIS_RESULT_BACKEND.build_redis_dsn(
-                    RedisDatabase.CELERY_TASKS
-                ),
-                client_name=f"{fastapi_app.title}.celery_tasks",
-            )
-
+        async def setup_task_manager():
             assert sender.app  # nosec
             assert isinstance(sender.app, Celery)  # nosec
             set_task_manager(
                 sender.app,
-                CeleryTaskManager(
+                create_task_manager(
                     sender.app,
                     celery_settings,
-                    RedisTaskInfoStore(redis_client_sdk),
                 ),
             )
 
@@ -83,7 +72,7 @@ async def fastapi_lifespan(
         set_event_loop(fastapi_app, loop)
 
         set_fastapi_app(sender.app, fastapi_app)
-        loop.run_until_complete(setup_task_worker())
+        loop.run_until_complete(setup_task_manager())
         loop.run_until_complete(
             fastapi_lifespan(startup_complete_event, shutdown_event)
         )