refactor: queue names

Author: giancarloromeo

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

@@ -82,7 +82,7 @@ async def rpc_sync_job(
     task_manager: TaskManager, *, job_id_data: AsyncJobNameData, **kwargs: Any
 ) -> AsyncJobGet:
     task_name = sync_job.__name__
-    task_uuid = await task_manager.submit_task(
+    task_uuid = await task_manager.send_task(
         TaskMetadata(name=task_name), task_context=job_id_data.model_dump(), **kwargs
     )
 
@@ -94,7 +94,7 @@ async def rpc_async_job(
     task_manager: TaskManager, *, job_id_data: AsyncJobNameData, **kwargs: Any
 ) -> AsyncJobGet:
     task_name = async_job.__name__
-    task_uuid = await task_manager.submit_task(
+    task_uuid = await task_manager.send_task(
         TaskMetadata(name=task_name), task_context=job_id_data.model_dump(), **kwargs
     )
 

packages/service-library/src/servicelib/celery/models.py

@@ -11,6 +11,17 @@
 TaskName: TypeAlias = Annotated[
     str, StringConstraints(strip_whitespace=True, min_length=1)
 ]
+
+TaskQueue: TypeAlias = Annotated[
+    str,
+    StringConstraints(
+        strip_whitespace=True,
+        min_length=1,
+        max_length=64,
+    ),
+]
+TASK_QUEUE_DEFAULT: TaskQueue = "default"
+
 TaskUUID: TypeAlias = UUID
 
 
@@ -23,15 +34,10 @@ class TaskState(StrEnum):
     ABORTED = "ABORTED"
 
 
-class TasksQueue(StrEnum):
-    CPU_BOUND = "cpu_bound"
-    DEFAULT = "default"
-
-
 class TaskMetadata(BaseModel):
     name: TaskName
     ephemeral: bool = True
-    queue: TasksQueue = TasksQueue.DEFAULT
+    queue: TaskQueue = TASK_QUEUE_DEFAULT
 
 
 class Task(BaseModel):

packages/service-library/src/servicelib/celery/task_manager.py

@@ -13,8 +13,8 @@
 
 
 class TaskManager(Protocol):
-    async def submit_task(
-        self, task_metadata: TaskMetadata, *, task_context: TaskContext, **task_param
+    async def send_task(
+        self, task_metadata: TaskMetadata, *, task_context: TaskContext, **task_params
     ) -> TaskUUID: ...
 
     async def cancel_task(

services/docker-compose.yml

@@ -1299,7 +1299,7 @@ services:
       STORAGE_WORKER_NAME: "sto-worker-cpu-bound-{{.Node.Hostname}}-{{.Task.Slot}}-{{.Task.ID}}"
       STORAGE_WORKER_MODE: "true"
       CELERY_CONCURRENCY: 1
-      CELERY_QUEUES: "cpu_bound"
+      CELERY_QUEUES: "storage.cpu_bound"
     networks: *storage_networks
 
   rabbit:

services/notifications/src/simcore_service_notifications/modules/celery/tasks.py

@@ -1,4 +1,5 @@
 import logging
+from enum import StrEnum
 
 from celery import Celery  # type: ignore[import-untyped]
 from celery_library.task import register_task
@@ -11,6 +12,13 @@
 _logger = logging.getLogger(__name__)
 
 
+_TASK_QUEUE_PREFIX: str = "notifications"
+
+
+class TaskQueue(StrEnum):
+    DEFAULT = f"{_TASK_QUEUE_PREFIX}.default"
+
+
 def setup_worker_tasks(app: Celery) -> None:
     register_celery_types()
     register_pydantic_types(NotificationMessage, EmailRecipient, SMSRecipient)

services/notifications/src/simcore_service_notifications/services/notifications_service.py

@@ -1,11 +1,30 @@
+from enum import StrEnum
+
+from servicelib.celery.models import TaskContext, TaskMetadata
 from servicelib.celery.task_manager import TaskManager
 
 from ..models.schemas import NotificationMessage, Recipient
 
+CHANNELS = {
+    "email": "notifications.email",
+}
+
+
+class TaskQueues(StrEnum):
+    DEFAULT = "notifications.default"
+
 
 async def send_notification(
     task_manager: TaskManager,
     *,
     message: NotificationMessage,
     recipients: list[Recipient],
-) -> None: ...
+) -> None:
+    for recipient in recipients:
+        await task_manager.send_task(
+            TaskMetadata(
+                name="notifications.send_email",
+                queue=TaskQueues.DEFAULT,
+            ),
+            task_context=TaskContext(),
+        )

services/storage/src/simcore_service_storage/api/rest/_files.py

@@ -288,7 +288,7 @@ async def complete_upload_file(
         user_id=query_params.user_id,
         product_name=_UNDEFINED_PRODUCT_NAME_FOR_WORKER_TASKS,  # NOTE: I would need to change the API here
     )
-    task_uuid = await task_manager.submit_task(
+    task_uuid = await task_manager.send_task(
         TaskMetadata(
             name=remote_complete_upload_file.__name__,
         ),

services/storage/src/simcore_service_storage/api/rpc/_paths.py

@@ -25,7 +25,7 @@ async def compute_path_size(
     path: Path,
 ) -> AsyncJobGet:
     task_name = remote_compute_path_size.__name__
-    task_uuid = await task_manager.submit_task(
+    task_uuid = await task_manager.send_task(
         task_metadata=TaskMetadata(
             name=task_name,
         ),
@@ -46,7 +46,7 @@ async def delete_paths(
     paths: set[Path],
 ) -> AsyncJobGet:
     task_name = remote_delete_paths.__name__
-    task_uuid = await task_manager.submit_task(
+    task_uuid = await task_manager.send_task(
         task_metadata=TaskMetadata(
             name=task_name,
         ),

services/storage/src/simcore_service_storage/api/rpc/_simcore_s3.py

@@ -4,10 +4,11 @@
 )
 from models_library.api_schemas_storage.storage_schemas import FoldersBody
 from models_library.api_schemas_webserver.storage import PathToExport
-from servicelib.celery.models import TaskMetadata, TasksQueue
+from servicelib.celery.models import TaskMetadata
 from servicelib.celery.task_manager import TaskManager
 from servicelib.rabbitmq import RPCRouter
 
+from ...modules.celery.tasks import TaskQueue
 from .._worker_tasks._simcore_s3 import deep_copy_files_from_project, export_data
 
 router = RPCRouter()
@@ -20,7 +21,7 @@ async def copy_folders_from_project(
     body: FoldersBody,
 ) -> AsyncJobGet:
     task_name = deep_copy_files_from_project.__name__
-    task_uuid = await task_manager.submit_task(
+    task_uuid = await task_manager.send_task(
         task_metadata=TaskMetadata(
             name=task_name,
         ),
@@ -39,11 +40,11 @@ async def start_export_data(
     paths_to_export: list[PathToExport],
 ) -> AsyncJobGet:
     task_name = export_data.__name__
-    task_uuid = await task_manager.submit_task(
+    task_uuid = await task_manager.send_task(
         task_metadata=TaskMetadata(
             name=task_name,
             ephemeral=False,
-            queue=TasksQueue.CPU_BOUND,
+            queue=TaskQueue.CPU_BOUND,
         ),
         task_context=job_id_data.model_dump(),
         user_id=job_id_data.user_id,

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

@@ -0,0 +1,8 @@
+from enum import StrEnum
+
+_TASK_QUEUE_PREFIX: str = "storage."
+
+
+class TaskQueue(StrEnum):
+    DEFAULT = f"{_TASK_QUEUE_PREFIX}.default"
+    CPU_BOUND = f"{_TASK_QUEUE_PREFIX}.cpu_bound"