fix(notifications): paginate and index the notification list endpoints (fork)

alembic/versions/b3d51c7fa9e2_index_notification_timestamp_and_read.py

@@ -0,0 +1,44 @@
+"""index notification timestamp and read
+
+Revision ID: b3d51c7fa9e2
+Revises: e60ae827ed98
+Create Date: 2026-07-11 10:12:38.114905
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision: str = 'b3d51c7fa9e2'
+down_revision: Union[str, None] = 'e60ae827ed98'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # The notification list is ordered by timestamp descending and paginated,
+    # so the index matches that order to keep a page from sorting the table.
+    op.create_index(
+        'ix_notification_timestamp_desc',
+        'notification',
+        [sa.text('timestamp DESC')],
+        unique=False,
+    )
+    # Unread notifications are a small fraction of the table, so a partial
+    # index keeps the unread query off a full table scan.
+    op.create_index(
+        'ix_notification_unread_timestamp_desc',
+        'notification',
+        [sa.text('timestamp DESC')],
+        unique=False,
+        postgresql_where=sa.text('read = false'),
+    )
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    op.drop_index('ix_notification_unread_timestamp_desc', table_name='notification')
+    op.drop_index('ix_notification_timestamp_desc', table_name='notification')

media_manager/main.py

@@ -141,6 +141,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator:
     allow_origins=origins,
     allow_credentials=True,
     allow_methods=["GET", "PUT", "POST", "DELETE", "PATCH", "HEAD", "OPTIONS"],
+    expose_headers=["X-Total-Count"],
 )
 app.add_middleware(CorrelationIdMiddleware, header_name="X-Correlation-ID")
 api_app = APIRouter(prefix="/api/v1")

media_manager/notification/models.py

@@ -1,6 +1,6 @@
 from uuid import UUID
 
-from sqlalchemy import DateTime
+from sqlalchemy import DateTime, Index, text
 from sqlalchemy.orm import Mapped, mapped_column
 
 from media_manager.database import Base
@@ -13,3 +13,17 @@ class Notification(Base):
     message: Mapped[str]
     read: Mapped[bool]
     timestamp = mapped_column(DateTime, nullable=False)
+
+    __table_args__ = (
+        # Notifications are always listed newest first and are paginated, so the
+        # index matches that order to keep a page from sorting the whole table.
+        Index("ix_notification_timestamp_desc", text("timestamp DESC")),
+        # Unread notifications are a small fraction of the table, so a partial
+        # index keeps the unread query off a full table scan.
+        Index(
+            "ix_notification_unread_timestamp_desc",
+            text("timestamp DESC"),
+            postgresql_where=text("read = false"),
+            sqlite_where=text("read = 0"),
+        ),
+    )

media_manager/notification/repository.py

@@ -1,12 +1,13 @@
 import logging
 
-from sqlalchemy import delete, select, update
+from sqlalchemy import delete, func, select, update
 from sqlalchemy.exc import (
     IntegrityError,
     SQLAlchemyError,
 )
 from sqlalchemy.ext.asyncio import AsyncSession
-from sqlalchemy.sql.expression import false
+from sqlalchemy.sql import Select
+from sqlalchemy.sql.expression import false, true
 
 from media_manager.exceptions import ConflictError, NotFoundError
 from media_manager.notification.models import Notification
@@ -20,6 +21,12 @@
 log = logging.getLogger(__name__)
 
 
+def _filtered_by_read(stmt: Select, read: bool | None) -> Select:
+    if read is None:
+        return stmt
+    return stmt.where(Notification.read == (true() if read else false()))
+
+
 class NotificationRepository:
     def __init__(self, db: AsyncSession) -> None:
         self.db = db
@@ -33,12 +40,15 @@ async def get_notification(self, nid: NotificationId) -> NotificationSchema:
 
         return NotificationSchema.model_validate(result)
 
-    async def get_unread_notifications(self) -> list[NotificationSchema]:
+    async def get_unread_notifications(
+        self, limit: int, offset: int = 0
+    ) -> list[NotificationSchema]:
         try:
             stmt = (
-                select(Notification)
-                .where(Notification.read == false())
+                _filtered_by_read(select(Notification), read=False)
                 .order_by(Notification.timestamp.desc())
+                .limit(limit)
+                .offset(offset)
             )
             results = (await self.db.execute(stmt)).scalars().all()
             return [
@@ -49,9 +59,21 @@ async def get_unread_notifications(self) -> list[NotificationSchema]:
             log.exception("Database error while retrieving unread notifications")
             raise
 
-    async def get_all_notifications(self) -> list[NotificationSchema]:
+    async def get_all_notifications(
+        self, limit: int, offset: int = 0, read: bool | None = None
+    ) -> list[NotificationSchema]:
+        """Return one page of notifications, newest first.
+
+        ``read`` optionally restricts the page to read or unread notifications;
+        ``None`` returns both.
+        """
         try:
-            stmt = select(Notification).order_by(Notification.timestamp.desc())
+            stmt = (
+                _filtered_by_read(select(Notification), read=read)
+                .order_by(Notification.timestamp.desc())
+                .limit(limit)
+                .offset(offset)
+            )
             results = (await self.db.execute(stmt)).scalars().all()
             return [
                 NotificationSchema.model_validate(notification)
@@ -61,6 +83,17 @@ async def get_all_notifications(self) -> list[NotificationSchema]:
             log.exception("Database error while retrieving notifications")
             raise
 
+    async def count_notifications(self, read: bool | None = None) -> int:
+        """Count the notifications a matching page would be taken from."""
+        try:
+            stmt = _filtered_by_read(
+                select(func.count()).select_from(Notification), read=read
+            )
+            return (await self.db.execute(stmt)).scalar_one()
+        except SQLAlchemyError:
+            log.exception("Database error while counting notifications")
+            raise
+
     async def save_notification(self, notification: NotificationSchema) -> None:
         try:
             self.db.add(
@@ -91,6 +124,18 @@ async def mark_notification_as_unread(self, nid: NotificationId) -> None:
         await self.db.execute(stmt)
         return
 
+    async def mark_all_notifications_as_read(self) -> None:
+        """Mark every unread notification as read in a single statement.
+
+        The UI only holds a page of notifications now, so it can no longer mark
+        them all read by looping over the ones it has.
+        """
+        stmt = (
+            update(Notification).where(Notification.read == false()).values(read=True)
+        )
+        await self.db.execute(stmt)
+        return
+
     async def delete_notification(self, nid: NotificationId) -> None:
         stmt = delete(Notification).where(Notification.id == nid)
         result = await self.db.execute(stmt)

media_manager/notification/router.py

@@ -1,11 +1,26 @@
-from fastapi import APIRouter, Depends, status
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Query, Response, status
 
 from media_manager.auth.users import current_active_user
 from media_manager.notification.dependencies import notification_service_dep
 from media_manager.notification.schemas import Notification, NotificationId
 
 router = APIRouter()
 
+DEFAULT_PAGE_SIZE = 50
+MAX_PAGE_SIZE = 200
+
+TOTAL_COUNT_HEADER = "X-Total-Count"
+
+limit_dep = Annotated[
+    int,
+    Query(
+        ge=1, le=MAX_PAGE_SIZE, description="Maximum number of notifications to return"
+    ),
+]
+offset_dep = Annotated[int, Query(ge=0, description="Number of notifications to skip")]
+
 
 # --------------------------------
 # GET NOTIFICATIONS
@@ -18,11 +33,25 @@
 )
 async def get_all_notifications(
     notification_service: notification_service_dep,
+    response: Response,
+    limit: limit_dep = DEFAULT_PAGE_SIZE,
+    offset: offset_dep = 0,
+    read: Annotated[
+        bool | None,
+        Query(description="Only return read (true) or unread (false) notifications"),
+    ] = None,
 ) -> list[Notification]:
     """
-    Get all notifications.
+    Get a page of notifications, newest first.
+
+    The total number of matching notifications is returned in the
+    `X-Total-Count` response header.
     """
-    return await notification_service.get_all_notifications()
+    total = await notification_service.count_notifications(read=read)
+    response.headers[TOTAL_COUNT_HEADER] = str(total)
+    return await notification_service.get_all_notifications(
+        limit=limit, offset=offset, read=read
+    )
 
 
 @router.get(
@@ -31,11 +60,21 @@ async def get_all_notifications(
 )
 async def get_unread_notifications(
     notification_service: notification_service_dep,
+    response: Response,
+    limit: limit_dep = DEFAULT_PAGE_SIZE,
+    offset: offset_dep = 0,
 ) -> list[Notification]:
     """
-    Get all unread notifications.
+    Get a page of unread notifications, newest first.
+
+    The total number of unread notifications is returned in the `X-Total-Count`
+    response header.
     """
-    return await notification_service.get_unread_notifications()
+    total = await notification_service.count_notifications(read=False)
+    response.headers[TOTAL_COUNT_HEADER] = str(total)
+    return await notification_service.get_unread_notifications(
+        limit=limit, offset=offset
+    )
 
 
 @router.get(
@@ -59,6 +98,23 @@ async def get_notification(
 # --------------------------------
 
 
+@router.patch(
+    "/read",
+    status_code=status.HTTP_204_NO_CONTENT,
+    dependencies=[Depends(current_active_user)],
+)
+async def mark_all_notifications_as_read(
+    notification_service: notification_service_dep,
+) -> None:
+    """
+    Mark every unread notification as read.
+
+    The client only holds a page of notifications, so it cannot mark them all
+    read by iterating over the ones it has.
+    """
+    await notification_service.mark_all_notifications_as_read()
+
+
 @router.patch(
     "/{notification_id}/read",
     status_code=status.HTTP_204_NO_CONTENT,

media_manager/notification/service.py

@@ -14,11 +14,22 @@ def __init__(
     async def get_notification(self, nid: NotificationId) -> Notification:
         return await self.notification_repository.get_notification(nid=nid)
 
-    async def get_unread_notifications(self) -> list[Notification]:
-        return await self.notification_repository.get_unread_notifications()
-
-    async def get_all_notifications(self) -> list[Notification]:
-        return await self.notification_repository.get_all_notifications()
+    async def get_unread_notifications(
+        self, limit: int, offset: int = 0
+    ) -> list[Notification]:
+        return await self.notification_repository.get_unread_notifications(
+            limit=limit, offset=offset
+        )
+
+    async def get_all_notifications(
+        self, limit: int, offset: int = 0, read: bool | None = None
+    ) -> list[Notification]:
+        return await self.notification_repository.get_all_notifications(
+            limit=limit, offset=offset, read=read
+        )
+
+    async def count_notifications(self, read: bool | None = None) -> int:
+        return await self.notification_repository.count_notifications(read=read)
 
     async def save_notification(self, notification: Notification) -> None:
         return await self.notification_repository.save_notification(notification)
@@ -29,6 +40,9 @@ async def mark_notification_as_read(self, nid: NotificationId) -> None:
     async def mark_notification_as_unread(self, nid: NotificationId) -> None:
         return await self.notification_repository.mark_notification_as_unread(nid=nid)
 
+    async def mark_all_notifications_as_read(self) -> None:
+        return await self.notification_repository.mark_all_notifications_as_read()
+
     async def delete_notification(self, nid: NotificationId) -> None:
         return await self.notification_repository.delete_notification(nid=nid)
 

tests/conftest.py

@@ -22,6 +22,7 @@
 from media_manager.movies import models as movie_models
 from media_manager.movies.schemas import Movie as MovieSchema
 from media_manager.movies.schemas import MovieId
+from media_manager.notification import models as notification_models
 from media_manager.tv import models as tv_models
 from media_manager.tv.schemas import Episode as EpisodeSchema
 from media_manager.tv.schemas import EpisodeId, SeasonId, ShowId
@@ -31,7 +32,7 @@
 # Importing the model modules is what registers their tables on ``Base.metadata``;
 # binding them here keeps the imports from looking unused. Note that the indexer
 # models are deliberately not imported: they use a PostgreSQL-only ARRAY column.
-_MODEL_MODULES = (movie_models, tv_models)
+_MODEL_MODULES = (movie_models, notification_models, tv_models)
 
 
 @pytest.fixture