Add notification endpoints

src/api/app.py

@@ -15,6 +15,7 @@ from .auth import router as auth_router
 from .categories import router as categories_router
 from .events import router as events_router
 from .invitations import router as invitations_router
+from .notifications import router as notifications_router
 from .sessions import router as sessions_router
 from .users import router as users_router
 
@@ -61,6 +62,7 @@ app.include_router(sessions_router)
 app.include_router(categories_router)
 app.include_router(events_router)
 app.include_router(invitations_router)
+app.include_router(notifications_router)
 
 
 @app.get("/ping")

src/api/notifications.py

@@ -0,0 +1,94 @@
+from datetime import datetime
+from typing import Any, Literal, cast, final
+
+from beanie import Link, PydanticObjectId
+from fastapi import APIRouter, HTTPException, status
+from pydantic import BaseModel
+
+from src.db.models.notificaton import Notification
+from src.db.models.user import User
+from src.utils.db import MissingIdError, UnfetchedLinkError, expr
+from src.utils.logging import get_logger
+
+from .auth import CurrentUserDep
+
+__all__ = ["router"]
+
+log = get_logger(__name__)
+
+base_router = APIRouter(tags=["Notifications"])
+notifications_router = APIRouter(tags=["Notifications"], prefix="/notifications")
+
+
+@final
+class NotificationData(BaseModel):
+    """Information about a notification sent to the user."""
+
+    id: PydanticObjectId
+    user_id: PydanticObjectId
+    event_type: Literal["reminder", "invitation"]
+    message: str
+    data: Any
+    read: bool
+    created_at: datetime
+    read_at: datetime | None
+
+    @classmethod
+    def from_notification(cls, notificaton: Notification) -> "NotificationData":
+        """Construct NotificationData from database Notification object."""
+        if notificaton.id is None:
+            raise MissingIdError(notificaton)
+
+        if isinstance(notificaton.user, Link):
+            raise UnfetchedLinkError(notificaton.user)
+        if notificaton.user.id is None:
+            raise MissingIdError(notificaton.user)
+
+        return cls(
+            id=notificaton.id,
+            user_id=notificaton.user.id,
+            event_type=notificaton.event_type,
+            message=notificaton.message,
+            data=notificaton.data,
+            read=notificaton.read,
+            created_at=notificaton.created_at,
+            read_at=notificaton.read_at,
+        )
+
+
+@base_router.get("/users/{user_id}/notifications")
+async def get_user_notifications(user_id: PydanticObjectId, user: CurrentUserDep) -> list[NotificationData]:
+    """Get all notifications for the user.
+
+    Note that this endpoint only allows you to access the notifications you received.
+    """
+    if user.id is None:
+        raise MissingIdError(user)
+
+    if user.id != user_id:
+        raise HTTPException(status.HTTP_403_FORBIDDEN, "You can only access your own notifications.")
+
+    notifications = await Notification.find(expr(Notification.user).id == user.id).to_list()
+    return [NotificationData.from_notification(notification) for notification in notifications]
+
+
+@notifications_router.get("{notification_id}")
+async def get_notification(notification_id: PydanticObjectId, user: CurrentUserDep) -> NotificationData:
+    """Get a single notification."""
+    notification = await Notification.get(notification_id, fetch_links=True)
+
+    if notification is None:
+        raise HTTPException(status.HTTP_404_NOT_FOUND, "Notification not found.")
+
+    if user.id is None:
+        raise MissingIdError(user)
+
+    if cast(User, notification.user).id != user.id:
+        raise HTTPException(status.HTTP_403_FORBIDDEN, "You can only access your own notifications.")
+
+    return NotificationData.from_notification(notification)
+
+
+router = APIRouter()
+router.include_router(base_router)
+router.include_router(notifications_router)