refactor models

Author: giancarloromeo

packages/celery-library/src/celery_library/backends/_redis.py

@@ -5,16 +5,16 @@
 
 from models_library.progress_bar import ProgressReport
 from pydantic import ValidationError
-from servicelib.redis._client import RedisClientSDK
-
-from ..models import (
+from servicelib.queued_tasks.models import (
     Task,
     TaskContext,
     TaskID,
     TaskMetadata,
     TaskUUID,
-    build_task_id_prefix,
 )
+from servicelib.redis._client import RedisClientSDK
+
+from ..utils import build_task_id_prefix
 
 _CELERY_TASK_INFO_PREFIX: Final[str] = "celery-task-info-"
 _CELERY_TASK_ID_KEY_ENCODING = "utf-8"

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

@@ -4,16 +4,12 @@
 
 from celery import Celery  # type: ignore[import-untyped]
 from celery.worker.worker import WorkController  # type: ignore[import-untyped]
-from servicelib.celery.app_server import STARTUP_TIMEOUT, BaseAppServer
 from servicelib.logging_utils import log_context
+from servicelib.queued_tasks.app_server import STARTUP_TIMEOUT, BaseAppServer
 from settings_library.celery import CelerySettings
 
 from .common import create_task_manager
-from .utils import (
-    get_app_server,
-    set_app_server,
-    set_task_manager,
-)
+from .utils import get_app_server, set_app_server
 
 _logger = logging.getLogger(__name__)
 
@@ -36,22 +32,19 @@ async def _setup_task_manager():
             assert sender.app  # nosec
             assert isinstance(sender.app, Celery)  # nosec
 
-            set_app_server(sender.app, app_server)
-            set_task_manager(
+            app_server.task_manager = await create_task_manager(
                 sender.app,
-                await create_task_manager(
-                    sender.app,
-                    celery_settings,
-                ),
+                celery_settings,
             )
 
-        async def _setup_app_server():
-            await app_server.startup(startup_complete_event, shutdown_event)
+            set_app_server(sender.app, app_server)
 
         app_server.event_loop = loop
 
         loop.run_until_complete(_setup_task_manager())
-        loop.run_until_complete(_setup_app_server())
+        loop.run_until_complete(
+            app_server.startup(startup_complete_event, shutdown_event)
+        )
 
     thread = threading.Thread(
         group=None,

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

@@ -14,9 +14,9 @@
 from celery.exceptions import Ignore  # type: ignore[import-untyped]
 from pydantic import NonNegativeInt
 from servicelib.async_utils import cancel_wait_task
+from servicelib.queued_tasks.models import TaskID
 
 from .errors import encode_celery_transferrable_error
-from .models import TaskID
 from .utils import get_app_server
 
 _logger = logging.getLogger(__name__)

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

@@ -10,9 +10,7 @@
 from common_library.async_tools import make_async
 from models_library.progress_bar import ProgressReport
 from servicelib.logging_utils import log_context
-from settings_library.celery import CelerySettings
-
-from .models import (
+from servicelib.queued_tasks.models import (
     Task,
     TaskContext,
     TaskID,
@@ -21,8 +19,10 @@
     TaskState,
     TaskStatus,
     TaskUUID,
-    build_task_id,
 )
+from settings_library.celery import CelerySettings
+
+from .utils import build_task_id
 
 _logger = logging.getLogger(__name__)
 

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

@@ -1,10 +1,24 @@
-from celery import Celery  # type: ignore[import-untyped]
-from servicelib.celery.app_server import BaseAppServer
+from typing import Final
 
-from .task_manager import CeleryTaskManager
+from celery import Celery  # type: ignore[import-untyped]
+from servicelib.queued_tasks.app_server import BaseAppServer
+from servicelib.queued_tasks.models import TaskContext, TaskID, TaskUUID
 
 _APP_SERVER_KEY = "app_server"
-_TASK_MANAGER_KEY = "task_manager"
+
+_TASK_ID_KEY_DELIMITATOR: Final[str] = ":"
+
+
+def build_task_id_prefix(task_context: TaskContext) -> str:
+    return _TASK_ID_KEY_DELIMITATOR.join(
+        [f"{task_context[key]}" for key in sorted(task_context)]
+    )
+
+
+def build_task_id(task_context: TaskContext, task_uuid: TaskUUID) -> TaskID:
+    return _TASK_ID_KEY_DELIMITATOR.join(
+        [build_task_id_prefix(task_context), f"{task_uuid}"]
+    )
 
 
 def get_app_server(app: Celery) -> BaseAppServer:
@@ -15,13 +29,3 @@ def get_app_server(app: Celery) -> BaseAppServer:
 
 def set_app_server(app: Celery, app_server: BaseAppServer) -> None:
     app.conf[_APP_SERVER_KEY] = app_server
-
-
-def get_task_manager(celery_app: Celery) -> CeleryTaskManager:
-    worker = celery_app.conf[_TASK_MANAGER_KEY]
-    assert isinstance(worker, CeleryTaskManager)
-    return worker
-
-
-def set_task_manager(celery_app: Celery, worker: CeleryTaskManager) -> None:
-    celery_app.conf[_TASK_MANAGER_KEY] = worker

packages/celery-library/tests/conftest.py

@@ -13,10 +13,10 @@
 from celery.worker.worker import WorkController
 from celery_library.common import create_task_manager
 from celery_library.signals import on_worker_init, on_worker_shutdown
-from celery_library.utils import CeleryTaskManager
+from celery_library.task_manager import CeleryTaskManager
 from pytest_simcore.helpers.monkeypatch_envs import setenvs_from_dict
 from pytest_simcore.helpers.typing_env import EnvVarsDict
-from servicelib.celery.app_server import BaseAppServer
+from servicelib.queued_tasks.app_server import BaseAppServer
 from settings_library.celery import CelerySettings
 from settings_library.redis import RedisSettings
 

packages/celery-library/tests/unit/test_tasks.py

@@ -14,21 +14,21 @@
 from celery import Celery, Task
 from celery.contrib.abortable import AbortableTask
 from celery_library.errors import TransferrableCeleryError
-from celery_library.models import (
-    TaskContext,
-    TaskID,
-    TaskMetadata,
-    TaskState,
-)
 from celery_library.task import (
     AbortableAsyncResult,
     register_task,
 )
 from celery_library.task_manager import CeleryTaskManager
-from celery_library.utils import get_app_server, get_task_manager
+from celery_library.utils import get_app_server
 from common_library.errors_classes import OsparcErrorMixin
 from models_library.progress_bar import ProgressReport
 from servicelib.logging_utils import log_context
+from servicelib.queued_tasks.models import (
+    TaskContext,
+    TaskID,
+    TaskMetadata,
+    TaskState,
+)
 from tenacity import Retrying, retry_if_exception_type, stop_after_delay, wait_fixed
 
 _logger = logging.getLogger(__name__)
@@ -40,14 +40,12 @@
 async def _fake_file_processor(
     celery_app: Celery, task_name: str, task_id: str, files: list[str]
 ) -> str:
-    worker = get_task_manager(celery_app)
-
     def sleep_for(seconds: float) -> None:
         time.sleep(seconds)
 
     for n, file in enumerate(files, start=1):
         with log_context(_logger, logging.INFO, msg=f"Processing file {file}"):
-            await worker.set_task_progress(
+            await get_app_server(celery_app).task_manager.set_task_progress(
                 task_id=task_id,
                 report=ProgressReport(actual_value=n / len(files)),
             )

packages/service-library/src/servicelib/celery/__init__.py renamed to packages/service-library/src/servicelib/fastapi/queued_tasks/__init__.py

@@ -4,7 +4,7 @@
 from asgi_lifespan import LifespanManager
 from fastapi import FastAPI
 
-from ...celery.app_server import BaseAppServer
+from ...queued_tasks.app_server import BaseAppServer
 
 _SHUTDOWN_TIMEOUT: Final[float] = timedelta(seconds=10).total_seconds()
 _STARTUP_TIMEOUT: Final[float] = timedelta(minutes=1).total_seconds()