Create notification table

h/migrations/versions/372308320143_create_notification_table.py

@@ -0,0 +1,63 @@
+"""Create the notification table.
+
+Revision ID: 372308320143
+Revises: 7f9cbef8bb18
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+from h.db import types
+
+revision = "372308320143"
+down_revision = "7f9cbef8bb18"
+
+
+def upgrade() -> None:
+    op.create_table(
+        "notification",
+        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column("annotation_id", types.URLSafeUUID(), nullable=False),
+        sa.Column("user_id", sa.Integer(), nullable=False),
+        sa.Column(
+            "type", sa.Enum("MENTION", "REPLY", name="notificationtype"), nullable=False
+        ),
+        sa.Column(
+            "created", sa.DateTime(), server_default=sa.text("now()"), nullable=False
+        ),
+        sa.Column(
+            "updated", sa.DateTime(), server_default=sa.text("now()"), nullable=False
+        ),
+        sa.ForeignKeyConstraint(
+            ["annotation_id"],
+            ["annotation.id"],
+            name=op.f("fk__notification__annotation_id__annotation"),
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["user_id"],
+            ["user.id"],
+            name=op.f("fk__notification__user_id__user"),
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk__notification")),
+        sa.UniqueConstraint(
+            "user_id", "annotation_id", name="uq__notification__user_id__annotation_id"
+        ),
+    )
+    op.create_index(
+        op.f("ix__notification_annotation_id"),
+        "notification",
+        ["annotation_id"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix__notification_user_id"), "notification", ["user_id"], unique=False
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(op.f("ix__notification_user_id"), table_name="notification")
+    op.drop_index(op.f("ix__notification_annotation_id"), table_name="notification")
+    op.drop_table("notification")
+    op.execute("DROP TYPE IF EXISTS notificationtype")

h/models/__init__.py

@@ -34,6 +34,7 @@
 from h.models.group_scope import GroupScope
 from h.models.job import Job
 from h.models.mention import Mention
+from h.models.notification import Notification
 from h.models.organization import Organization
 from h.models.setting import Setting
 from h.models.subscriptions import Subscriptions
@@ -63,6 +64,7 @@
     "GroupScope",
     "Job",
     "Mention",
+    "Notification",
     "Organization",
     "Setting",
     "Subscriptions",

h/models/annotation.py

@@ -140,6 +140,8 @@ class Annotation(Base):
 
     mentions = sa.orm.relationship("Mention", back_populates="annotation")
 
+    notifications = sa.orm.relationship("Notification", back_populates="annotation")
+
     @property
     def uuid(self):
         """

h/models/notification.py

@@ -0,0 +1,55 @@
+import enum
+
+from sqlalchemy import (
+    ForeignKey,
+    UniqueConstraint,
+)
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from h.db import Base, types
+from h.db.mixins import Timestamps
+from h.models import helpers
+
+
+class NotificationType(enum.StrEnum):
+    MENTION = "mention"
+    REPLY = "reply"
+
+
+class Notification(Base, Timestamps):
+    __tablename__ = "notification"
+
+    id: Mapped[int] = mapped_column(autoincrement=True, primary_key=True)
+
+    annotation_id: Mapped[types.URLSafeUUID] = mapped_column(
+        ForeignKey("annotation.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    """FK to annotation.id"""
+    annotation = relationship(
+        "Annotation", back_populates="notifications", uselist=False
+    )
+
+    user_id: Mapped[int] = mapped_column(
+        ForeignKey("user.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    """FK to user.id"""
+    user = relationship("User", uselist=False)
+
+    type: Mapped[NotificationType]
+    """Notification type"""
+
+    __table_args__ = (
+        # Ensure that a user can only have one notification for a given annotation
+        UniqueConstraint(
+            "user_id",
+            "annotation_id",
+            name="uq__notification__user_id__annotation_id",
+        ),
+    )
+
+    def __repr__(self) -> str:
+        return helpers.repr_(self, ["id", "annotation_id"])