feat(): push notifications via firebase admin

src/template_project/application/common/notifications/service.py

@@ -0,0 +1,8 @@
+from abc import abstractmethod
+from typing import Protocol
+
+
+class NotificationService(Protocol):
+    @abstractmethod
+    async def send_notification(self, identifier: str, title: str, body: str) -> None:
+        raise NotImplementedError

src/template_project/application/user/notification/interactors/send_notification.py

@@ -0,0 +1,32 @@
+from template_project.application.common.data_structure import to_data_structure
+from template_project.application.common.identity_provider import IdentityProvider
+from template_project.application.common.interactor import to_interactor
+from template_project.application.common.notifications.service import NotificationService
+from template_project.application.user.notification_device.data_gateway import NotificationDeviceDataGateway
+from template_project.application.user.notification_device.errors import NotificationDeviceNotFoundError
+
+
+@to_data_structure
+class SendNotificationRequest:
+    title: str
+    body: str
+
+
+@to_interactor
+class NotificationInteractor:
+    identity_provider: IdentityProvider
+    notification_device_data_gateway: NotificationDeviceDataGateway
+    notification_service: NotificationService
+
+    async def send_notification(self, request: SendNotificationRequest) -> None:
+        current_user = await self.identity_provider.get_current_user()
+        notification_device = await self.notification_device_data_gateway.load_by_user_id(current_user.id)
+
+        if not notification_device:
+            raise NotificationDeviceNotFoundError
+
+        await self.notification_service.send_notification(
+            identifier=notification_device.device_id,
+            title=request.title,
+            body=request.body,
+        )

src/template_project/application/user/notification_device/data_gateway.py

@@ -0,0 +1,11 @@
+from abc import abstractmethod
+from typing import Protocol
+
+from template_project.application.user.entity import UserId
+from template_project.application.user.notification_device.entity import NotificationDevice
+
+
+class NotificationDeviceDataGateway(Protocol):
+    @abstractmethod
+    async def load_by_user_id(self, user_id: UserId) -> NotificationDevice | None:
+        raise NotImplementedError

src/template_project/application/user/notification_device/entity.py

@@ -0,0 +1,29 @@
+from datetime import UTC, datetime
+from typing import NewType, Self
+from uuid import UUID
+
+from uuid_utils.compat import uuid7
+
+from template_project.application.common.entity import Entity, to_entity
+from template_project.application.user.entity import UserId
+
+NotificationDeviceId = NewType("NotificationDeviceId", UUID)
+
+
+@to_entity
+class NotificationDevice(Entity[NotificationDeviceId]):
+    user_id: UserId
+    device_id: str
+
+    @classmethod
+    def factory(
+        cls,
+        user_id: UserId,
+        device_id: str,
+    ) -> Self:
+        return cls(
+            id=NotificationDeviceId(uuid7()),
+            user_id=user_id,
+            device_id=device_id,
+            created_at=datetime.now(tz=UTC),
+        )

src/template_project/application/user/notification_device/errors.py

@@ -0,0 +1,6 @@
+from template_project.application.common.errors import ApplicationError, to_error
+
+
+@to_error
+class NotificationDeviceNotFoundError(ApplicationError):
+    pass