WIP: ✨ Add Notifications service implementation

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

@@ -10,11 +10,14 @@
 from common_library.async_tools import make_async
 from models_library.progress_bar import ProgressReport
 from servicelib.celery.models import (
+    TASK_QUEUE_DEFAULT,
     Task,
     TaskFilter,
     TaskID,
     TaskInfoStore,
     TaskMetadata,
+    TaskName,
+    TaskQueue,
     TaskState,
     TaskStatus,
     TaskUUID,
@@ -38,34 +41,43 @@ class CeleryTaskManager:
     _celery_settings: CelerySettings
     _task_info_store: TaskInfoStore
 
-    async def submit_task(
+    async def send_task(
         self,
-        task_metadata: TaskMetadata,
-        *,
+        task_name: TaskName,
         task_filter: TaskFilter,
+        *,
+        task_ephemeral: bool = True,
+        task_queue: TaskQueue = TASK_QUEUE_DEFAULT,
         **task_params,
     ) -> TaskUUID:
         with log_context(
             _logger,
             logging.DEBUG,
-            msg=f"Submit {task_metadata.name=}: {task_filter=} {task_params=}",
+            msg=f"Send {task_name=}: {task_filter=} {task_params=}",
         ):
             task_uuid = uuid4()
             task_id = build_task_id(task_filter, task_uuid)
             self._celery_app.send_task(
-                task_metadata.name,
+                task_name,
                 task_id=task_id,
                 kwargs={"task_id": task_id} | task_params,
-                queue=task_metadata.queue.value,
+                queue=task_queue,
             )
 
             expiry = (
                 self._celery_settings.CELERY_EPHEMERAL_RESULT_EXPIRES
-                if task_metadata.ephemeral
+                if task_ephemeral
                 else self._celery_settings.CELERY_RESULT_EXPIRES
             )
+
             await self._task_info_store.create_task(
-                task_id, task_metadata, expiry=expiry
+                task_id,
+                TaskMetadata(
+                    name=task_name,
+                    ephemeral=task_ephemeral,
+                    queue=task_queue,
+                ),
+                expiry=expiry,
             )
             return task_uuid
 
@@ -97,6 +109,7 @@ async def get_task_result(
             msg=f"Get task result: {task_filter=} {task_uuid=}",
         ):
             task_id = build_task_id(task_filter, task_uuid)
+
             async_result = self._celery_app.AsyncResult(task_id)
             result = async_result.result
             if async_result.ready():
@@ -106,7 +119,7 @@ async def get_task_result(
                     await self._task_info_store.remove_task(task_id)
             return result
 
-    async def _get_task_progress_report(
+    async def _get_progress_report(
         self, task_filter: TaskFilter, task_uuid: TaskUUID, task_state: TaskState
     ) -> ProgressReport:
         if task_state in (TaskState.STARTED, TaskState.RETRY, TaskState.ABORTED):
@@ -144,7 +157,7 @@ async def get_task_status(
             return TaskStatus(
                 task_uuid=task_uuid,
                 task_state=task_state,
-                progress_report=await self._get_task_progress_report(
+                progress_report=await self._get_progress_report(
                     task_filter, task_uuid, task_state
                 ),
             )

packages/celery-library/tests/conftest.py

@@ -121,7 +121,7 @@ def _on_worker_init_wrapper(sender: WorkController, **_kwargs):
 
 
 @pytest.fixture
-async def celery_task_manager(
+async def task_manager(
     celery_app: Celery,
     celery_settings: CelerySettings,
     with_celery_worker: TestWorkController,

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

@@ -27,7 +27,7 @@
 from models_library.rabbitmq_basic_types import RPCNamespace
 from models_library.users import UserID
 from pydantic import TypeAdapter
-from servicelib.celery.models import TaskFilter, TaskID, TaskMetadata
+from servicelib.celery.models import TaskFilter, TaskID
 from servicelib.celery.task_manager import TaskManager
 from servicelib.rabbitmq import RabbitMQRPCClient, RPCRouter
 from servicelib.rabbitmq.rpc_interfaces.async_jobs import async_jobs
@@ -83,8 +83,8 @@ async def rpc_sync_job(
 ) -> AsyncJobGet:
     task_name = sync_job.__name__
     task_filter = TaskFilter.model_validate(job_filter.model_dump())
-    task_uuid = await task_manager.submit_task(
-        TaskMetadata(name=task_name), task_filter=task_filter, **kwargs
+    task_uuid = await task_manager.send_task(
+        task_name=task_name, task_filter=task_filter, **kwargs
     )
 
     return AsyncJobGet(job_id=task_uuid, job_name=task_name)
@@ -96,8 +96,8 @@ async def rpc_async_job(
 ) -> AsyncJobGet:
     task_name = async_job.__name__
     task_filter = TaskFilter.model_validate(job_filter.model_dump())
-    task_uuid = await task_manager.submit_task(
-        TaskMetadata(name=task_name), task_filter=task_filter, **kwargs
+    task_uuid = await task_manager.send_task(
+        task_name=task_name, task_filter=task_filter, **kwargs
     )
 
     return AsyncJobGet(job_id=task_uuid, job_name=task_name)
@@ -141,13 +141,13 @@ async def async_job(task: Task, task_id: TaskID, action: Action, payload: Any) -
 
 @pytest.fixture
 async def register_rpc_routes(
-    async_jobs_rabbitmq_rpc_client: RabbitMQRPCClient, celery_task_manager: TaskManager
+    async_jobs_rabbitmq_rpc_client: RabbitMQRPCClient, task_manager: TaskManager
 ) -> None:
     await async_jobs_rabbitmq_rpc_client.register_router(
-        _async_jobs.router, ASYNC_JOBS_RPC_NAMESPACE, task_manager=celery_task_manager
+        _async_jobs.router, ASYNC_JOBS_RPC_NAMESPACE, task_manager=task_manager
     )
     await async_jobs_rabbitmq_rpc_client.register_router(
-        router, ASYNC_JOBS_RPC_NAMESPACE, task_manager=celery_task_manager
+        router, ASYNC_JOBS_RPC_NAMESPACE, task_manager=task_manager
     )
 
 

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

@@ -22,7 +22,6 @@
 from servicelib.celery.models import (
     TaskFilter,
     TaskID,
-    TaskMetadata,
     TaskState,
 )
 from servicelib.logging_utils import log_context
@@ -91,14 +90,12 @@ def _(celery_app: Celery) -> None:
 
 
 async def test_submitting_task_calling_async_function_results_with_success_state(
-    celery_task_manager: CeleryTaskManager,
+    task_manager: CeleryTaskManager,
 ):
     task_filter = TaskFilter(user_id=42)
 
-    task_uuid = await celery_task_manager.submit_task(
-        TaskMetadata(
-            name=fake_file_processor.__name__,
-        ),
+    task_uuid = await task_manager.send_task(
+        task_name=fake_file_processor.__name__,
         task_filter=task_filter,
         files=[f"file{n}" for n in range(5)],
     )
@@ -109,26 +106,22 @@ async def test_submitting_task_calling_async_function_results_with_success_state
         stop=stop_after_delay(30),
     ):
         with attempt:
-            status = await celery_task_manager.get_task_status(task_filter, task_uuid)
+            status = await task_manager.get_task_status(task_filter, task_uuid)
             assert status.task_state == TaskState.SUCCESS
 
     assert (
-        await celery_task_manager.get_task_status(task_filter, task_uuid)
+        await task_manager.get_task_status(task_filter, task_uuid)
     ).task_state == TaskState.SUCCESS
-    assert (
-        await celery_task_manager.get_task_result(task_filter, task_uuid)
-    ) == "archive.zip"
+    assert (await task_manager.get_task_result(task_filter, task_uuid)) == "archive.zip"
 
 
 async def test_submitting_task_with_failure_results_with_error(
-    celery_task_manager: CeleryTaskManager,
+    task_manager: CeleryTaskManager,
 ):
     task_filter = TaskFilter(user_id=42)
 
-    task_uuid = await celery_task_manager.submit_task(
-        TaskMetadata(
-            name=failure_task.__name__,
-        ),
+    task_uuid = await task_manager.send_task(
+        task_name=failure_task.__name__,
         task_filter=task_filter,
     )
 
@@ -139,56 +132,50 @@ async def test_submitting_task_with_failure_results_with_error(
     ):
 
         with attempt:
-            raw_result = await celery_task_manager.get_task_result(
-                task_filter, task_uuid
-            )
+            raw_result = await task_manager.get_task_result(task_filter, task_uuid)
             assert isinstance(raw_result, TransferrableCeleryError)
 
-    raw_result = await celery_task_manager.get_task_result(task_filter, task_uuid)
+    raw_result = await task_manager.get_task_result(task_filter, task_uuid)
     assert f"{raw_result}" == "Something strange happened: BOOM!"
 
 
 async def test_cancelling_a_running_task_aborts_and_deletes(
-    celery_task_manager: CeleryTaskManager,
+    task_manager: CeleryTaskManager,
 ):
     task_filter = TaskFilter(user_id=42)
 
-    task_uuid = await celery_task_manager.submit_task(
-        TaskMetadata(
-            name=dreamer_task.__name__,
-        ),
+    task_uuid = await task_manager.send_task(
+        task_name=dreamer_task.__name__,
         task_filter=task_filter,
     )
 
     await asyncio.sleep(3.0)
 
-    await celery_task_manager.cancel_task(task_filter, task_uuid)
+    await task_manager.cancel_task(task_filter, task_uuid)
 
     for attempt in Retrying(
         retry=retry_if_exception_type(AssertionError),
         wait=wait_fixed(1),
         stop=stop_after_delay(30),
     ):
         with attempt:
-            progress = await celery_task_manager.get_task_status(task_filter, task_uuid)
+            progress = await task_manager.get_task_status(task_filter, task_uuid)
             assert progress.task_state == TaskState.ABORTED
 
     assert (
-        await celery_task_manager.get_task_status(task_filter, task_uuid)
+        await task_manager.get_task_status(task_filter, task_uuid)
     ).task_state == TaskState.ABORTED
 
-    assert task_uuid not in await celery_task_manager.list_tasks(task_filter)
+    assert task_uuid not in await task_manager.list_tasks(task_filter)
 
 
 async def test_listing_task_uuids_contains_submitted_task(
-    celery_task_manager: CeleryTaskManager,
+    task_manager: CeleryTaskManager,
 ):
     task_filter = TaskFilter(user_id=42)
 
-    task_uuid = await celery_task_manager.submit_task(
-        TaskMetadata(
-            name=dreamer_task.__name__,
-        ),
+    task_uuid = await task_manager.send_task(
+        task_name=dreamer_task.__name__,
         task_filter=task_filter,
     )
 
@@ -198,8 +185,8 @@ async def test_listing_task_uuids_contains_submitted_task(
         stop=stop_after_delay(10),
     ):
         with attempt:
-            tasks = await celery_task_manager.list_tasks(task_filter)
+            tasks = await task_manager.list_tasks(task_filter)
             assert any(task.uuid == task_uuid for task in tasks)
 
-        tasks = await celery_task_manager.list_tasks(task_filter)
+        tasks = await task_manager.list_tasks(task_filter)
         assert any(task.uuid == task_uuid for task in tasks)

packages/common-library/src/common_library/pydantic_basic_types.py

@@ -1,7 +1,7 @@
 from re import Pattern
 from typing import Annotated, Final, TypeAlias
 
-from pydantic import Field
+from pydantic import Field, StringConstraints
 from pydantic_core import core_schema
 
 # https://en.wikipedia.org/wiki/List_of_TCP_and_UDP_port_numbers#Registered_ports
@@ -77,3 +77,12 @@ class LongTruncatedStr(ConstrainedStr):
     # Analogous to ShortTruncatedStr
     strip_whitespace = True
     curtail_length = 65536  # same as github descripton
+
+
+NotEmptyStr: TypeAlias = Annotated[
+    str,
+    StringConstraints(
+        min_length=1,
+        strip_whitespace=True,
+    ),
+]

packages/models-library/requirements/_base.in

@@ -7,6 +7,7 @@
 arrow
 jsonschema
 orjson
-pydantic-extra-types
-pydantic-settings
+phonenumbers
 pydantic[email]
+pydantic-settings
+pydantic-extra-types

packages/models-library/requirements/_base.txt

@@ -22,6 +22,8 @@ orjson==3.10.15
     #   -c requirements/../../../requirements/constraints.txt
     #   -r requirements/../../../packages/common-library/requirements/_base.in
     #   -r requirements/_base.in
+phonenumbers==9.0.9
+    # via -r requirements/_base.in
 pydantic==2.11.7
     # via
     #   -c requirements/../../../packages/common-library/requirements/../../../requirements/constraints.txt

packages/models-library/src/models_library/rpc/notifications/__init__.py

@@ -0,0 +1,5 @@
+from ._notifications import Notification
+
+__all__: tuple[str, ...] = ("Notification",)
+
+# nopycln: file

packages/models-library/src/models_library/rpc/notifications/_notifications.py

@@ -0,0 +1,11 @@
+from typing import Annotated
+
+from pydantic import BaseModel, Field
+
+from .channels import Channel
+from .events import Event
+
+
+class Notification(BaseModel):
+    event: Annotated[Event, Field(discriminator="type")]
+    channel: Annotated[Channel, Field(discriminator="type")]

packages/models-library/src/models_library/rpc/notifications/channels/__init__.py

@@ -0,0 +1,16 @@
+from typing import TypeAlias
+
+from ._email_channel import EmailAddress, EmailChannel
+from ._sms_channel import SMSChannel
+
+Channel: TypeAlias = EmailChannel | SMSChannel
+
+
+__all__: tuple[str, ...] = (
+    "Channel",
+    "EmailAddress",
+    "EmailChannel",
+    "SMSChannel",
+)
+
+# nopycln: file

packages/models-library/src/models_library/rpc/notifications/channels/_email_channel.py

@@ -0,0 +1,16 @@
+from typing import Literal
+
+from pydantic import BaseModel, EmailStr
+
+
+class EmailAddress(BaseModel):
+    display_name: str = ""
+    addr_spec: EmailStr
+
+
+class EmailChannel(BaseModel):
+    type: Literal["email"] = "email"
+
+    from_addr: EmailAddress
+    to_addr: EmailAddress
+    reply_to_addr: EmailAddress | None = None

packages/models-library/src/models_library/rpc/notifications/channels/_sms_channel.py

@@ -0,0 +1,10 @@
+from typing import Literal
+
+from pydantic import BaseModel
+from pydantic_extra_types.phone_numbers import PhoneNumber
+
+
+class SMSChannel(BaseModel):
+    type: Literal["sms"] = "sms"
+
+    phone_number: PhoneNumber