fix: startup

Author: giancarloromeo

services/notifications/src/simcore_service_notifications/api/rpc/_notifications.py

@@ -1,16 +1,19 @@
-from fastapi import FastAPI
+from servicelib.celery.task_manager import TaskManager
 from servicelib.rabbitmq import RPCRouter
 
-from ...models.schemas import NotificationMessage
+from ...models.schemas import NotificationMessage, Recipient
 from ...services import notifications_service
 
 router = RPCRouter()
 
 
 @router.expose(reraise_if_error_type=())
 async def send_notification_message(
-    app: FastAPI,
+    task_manager: TaskManager,
     *,
     message: NotificationMessage,
+    recipients: list[Recipient],
 ) -> None:
-    await notifications_service.send_notification_message(message=message)
+    await notifications_service.send_notification(
+        task_manager, message=message, recipients=recipients
+    )

services/notifications/src/simcore_service_notifications/api/rpc/routes.py

@@ -5,6 +5,7 @@
 from models_library.api_schemas_notifications import NOTIFICATIONS_RPC_NAMESPACE
 from servicelib.rabbitmq import RPCRouter
 
+from ...clients.celery import get_task_manager_from_app
 from ...clients.rabbitmq import get_rabbitmq_rpc_server
 from . import _notifications
 
@@ -13,10 +14,10 @@
 
 async def rpc_api_routes_lifespan(app: FastAPI) -> AsyncIterator[State]:
     rpc_server = get_rabbitmq_rpc_server(app)
-
+    task_manager = get_task_manager_from_app(app)
     for router in ROUTERS:
         await rpc_server.register_router(
-            router, NOTIFICATIONS_RPC_NAMESPACE, app
+            router, NOTIFICATIONS_RPC_NAMESPACE, task_manager=task_manager
         )  # pragma: no cover
 
     yield {}

services/notifications/src/simcore_service_notifications/clients/celery.py

@@ -0,0 +1,35 @@
+import logging
+from collections.abc import AsyncIterator
+
+from celery_library.common import create_app, create_task_manager
+from celery_library.task_manager import CeleryTaskManager
+from celery_library.types import register_celery_types, register_pydantic_types
+from fastapi import FastAPI
+from fastapi_lifespan_manager import State
+from settings_library.celery import CelerySettings
+
+from ..core.settings import ApplicationSettings
+from ..models.schemas import EmailRecipient, NotificationMessage, SMSRecipient
+
+_logger = logging.getLogger(__name__)
+
+
+async def celery_lifespan(app: FastAPI) -> AsyncIterator[State]:
+    settings: ApplicationSettings = app.state.settings
+    if settings.NOTIFICATIONS_CELERY and not settings.NOTIFICATIONS_WORKER_MODE:
+        celery_settings: CelerySettings = settings.NOTIFICATIONS_CELERY
+
+        app.state.task_manager = await create_task_manager(
+            create_app(celery_settings), celery_settings
+        )
+
+        register_celery_types()
+        register_pydantic_types(NotificationMessage, EmailRecipient, SMSRecipient)
+    yield {}
+
+
+def get_task_manager_from_app(app: FastAPI) -> CeleryTaskManager:
+    assert hasattr(app.state, "task_manager")  # nosec
+    task_manager = app.state.task_manager
+    assert isinstance(task_manager, CeleryTaskManager)  # nosec
+    return task_manager

services/notifications/src/simcore_service_notifications/clients/rabbitmq.py

@@ -17,7 +17,7 @@ async def rabbitmq_lifespan(app: FastAPI) -> AsyncIterator[State]:
     await wait_till_rabbitmq_responsive(rabbit_settings.dsn)
 
     app.state.rabbitmq_rpc_server = await RabbitMQRPCClient.create(
-        client_name="dynamic_scheduler_rpc_server", settings=rabbit_settings
+        client_name="notifications_rpc_server", settings=rabbit_settings
     )
 
     yield {}

services/notifications/src/simcore_service_notifications/core/events.py

@@ -13,6 +13,7 @@
 
 from .._meta import APP_FINISHED_BANNER_MSG, APP_STARTED_BANNER_MSG
 from ..api.rpc.routes import rpc_api_routes_lifespan
+from ..clients.celery import celery_lifespan
 from ..clients.postgres import postgres_lifespan
 from ..clients.rabbitmq import rabbitmq_lifespan
 from .settings import ApplicationSettings
@@ -47,6 +48,9 @@ def create_app_lifespan():
     # - rabbitmq
     app_lifespan.add(rabbitmq_lifespan)
 
+    # - celery
+    app_lifespan.add(celery_lifespan)
+
     # - rpc api routes
     app_lifespan.add(rpc_api_routes_lifespan)
 

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

@@ -1,7 +1,17 @@
+import logging
+
+from celery import Task  # type: ignore[import-untyped]
+
 from ...models.schemas import EmailRecipient, NotificationMessage
 
+_logger = logging.getLogger(__name__)
+
 
-async def send_email_notification(
-    message: NotificationMessage, recipient: EmailRecipient
+async def send_email(
+    task: Task,
+    message: NotificationMessage,
+    recipient: EmailRecipient,
 ) -> None:
-    pass
+    # TODO: render email template with message and recipient details
+    #       and send the email using an email service
+    _logger.info(f"Sending email notification to {recipient.email}")

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

@@ -1,18 +1,19 @@
 import logging
 
 from celery import Celery  # type: ignore[import-untyped]
-from celery_library.types import register_celery_types
+from celery_library.task import register_task
+from celery_library.types import register_celery_types, register_pydantic_types
 from servicelib.logging_utils import log_context
 
+from ...models.schemas import NotificationMessage, SMSRecipient
+from ...modules.celery._email_tasks import EmailRecipient, send_email
+
 _logger = logging.getLogger(__name__)
 
 
 def setup_worker_tasks(app: Celery) -> None:
     register_celery_types()
-    # TODO: add more types as needed
-    # register_pydantic_types(FileUploadCompletionBody, FileMetaData, FoldersBody)
+    register_pydantic_types(NotificationMessage, EmailRecipient, SMSRecipient)
 
     with log_context(_logger, logging.INFO, msg="worker tasks registration"):
-        ...
-        # TODO: register tasks here
-        # register_task(app, send_email_notification)
+        register_task(app, send_email)

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

@@ -1,6 +1,11 @@
+from servicelib.celery.task_manager import TaskManager
+
 from ..models.schemas import NotificationMessage, Recipient
 
 
 async def send_notification(
-    message: NotificationMessage, *recipients: list[Recipient]
+    task_manager: TaskManager,
+    *,
+    message: NotificationMessage,
+    recipients: list[Recipient],
 ) -> None: ...