TO-291: Add notifications model

backend/migrations/versions/2026_03_10_1618-983168a63271_add_notifications.py

@@ -0,0 +1,55 @@
+# Copyright 2026 Canonical Ltd.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+#
+# SPDX-FileCopyrightText: Copyright 2026 Canonical Ltd.
+# SPDX-License-Identifier: AGPL-3.0-only
+
+"""Add notifications
+
+Revision ID: 983168a63271
+Revises: 3514f071a2e5
+Create Date: 2026-03-10 16:18:40.152277+00:00
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision = '983168a63271'
+down_revision = '3514f071a2e5'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('notification',
+    sa.Column('user_id', sa.Integer(), nullable=False),
+    sa.Column('notification_type', sa.Enum('USER_ASSIGNED_ARTEFACT_REVIEW', 'USER_ASSIGNED_ENVIRONMENT_REVIEW', name='notificationtype'), nullable=False),
+    sa.Column('target_url', sa.String(), nullable=True),
+    sa.Column('dismissed_at', sa.DateTime(), nullable=True),
+    sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+    sa.Column('created_at', sa.DateTime(), nullable=False),
+    sa.Column('updated_at', sa.DateTime(), nullable=False),
+    sa.ForeignKeyConstraint(['user_id'], ['app_user.id'], name=op.f('notification_user_id_fkey'), ondelete='CASCADE'),
+    sa.PrimaryKeyConstraint('id', name=op.f('notification_pkey'))
+    )
+    op.create_index(op.f('notification_user_id_ix'), 'notification', ['user_id'], unique=False)
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index(op.f('notification_user_id_ix'), table_name='notification')
+    op.drop_table('notification')
+    # ### end Alembic commands ###

backend/test_observer/controllers/notifications/models.py

@@ -0,0 +1,35 @@
+# Copyright 2024 Canonical Ltd.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License version 3, as
+# published by the Free Software Foundation.
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU Affero General Public License for more details.
+# You should have received a copy of the GNU Affero General Public License
+# along with this program.  If not, see <https://www.gnu.org/licenses/>.
+#
+# SPDX-FileCopyrightText: Copyright 2024 Canonical Ltd.
+# SPDX-License-Identifier: AGPL-3.0-only
+
+from datetime import datetime
+
+from pydantic import BaseModel, ConfigDict
+
+from test_observer.data_access.models_enums import NotificationType
+
+
+class NotificationResponse(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+
+    id: int
+    user_id: int
+    notification_type: NotificationType
+    target_url: str | None
+    created_at: datetime
+    dismissed_at: datetime | None
+
+
+class NotificationsResponse(BaseModel):
+    notifications: list[NotificationResponse]

backend/test_observer/data_access/models.py

@@ -51,6 +51,7 @@
     FamilyName,
     IssueSource,
     IssueStatus,
+    NotificationType,
     TestExecutionStatus,
     TestResultStatus,
 )
@@ -224,6 +225,20 @@ class UserSession(Base):
     user: Mapped[User] = relationship(back_populates="sessions", foreign_keys=[user_id])
 
 
+class Notification(Base):
+    __tablename__ = "notification"
+
+    user_id: Mapped[int] = mapped_column(ForeignKey("app_user.id", ondelete="CASCADE"), index=True)
+    notification_type: Mapped[NotificationType]
+    target_url: Mapped[str | None] = mapped_column(default=None)
+    dismissed_at: Mapped[datetime | None] = mapped_column(default=None)
+
+    user: Mapped[User] = relationship(foreign_keys=[user_id])
+
+    def __repr__(self) -> str:
+        return data_model_repr(self, "user_id", "notification_type")
+
+
 class Artefact(Base):
     """A model to represent artefacts (snaps, debs, images)"""
 

backend/test_observer/data_access/models_enums.py

@@ -118,3 +118,8 @@ class IssueStatus(StrEnum):
     UNKNOWN = "unknown"
     OPEN = "open"
     CLOSED = "closed"
+
+
+class NotificationType(StrEnum):
+    USER_ASSIGNED_ARTEFACT_REVIEW = "USER_ASSIGNED_ARTEFACT_REVIEW"
+    USER_ASSIGNED_ENVIRONMENT_REVIEW = "USER_ASSIGNED_ENVIRONMENT_REVIEW"