Feat/model checkpoints

h/migrations/versions/3794945d8e88_create_the_checkpoint_table.py

@@ -0,0 +1,54 @@
+"""Create the checkpoint table."""
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "3794945d8e88"
+down_revision = "a1b2c3d4e5f6"
+
+
+def upgrade() -> None:
+    op.create_table(
+        "checkpoint",
+        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column("group_id", sa.Integer(), nullable=False),
+        sa.Column("document_id", sa.Integer(), nullable=False),
+        sa.Column("previous_checkpoint_id", sa.Integer(), nullable=True),
+        sa.Column("reveal_date", sa.DateTime(), nullable=True),
+        sa.Column(
+            "updated", sa.DateTime(), server_default=sa.text("now()"), nullable=False
+        ),
+        sa.Column(
+            "created", sa.DateTime(), server_default=sa.text("now()"), nullable=False
+        ),
+        sa.ForeignKeyConstraint(
+            ["group_id"],
+            ["group.id"],
+            name=op.f("fk__checkpoint__group_id__group"),
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["document_id"],
+            ["document.id"],
+            name=op.f("fk__checkpoint__document_id__document"),
+            ondelete="CASCADE",
+        ),
+        sa.ForeignKeyConstraint(
+            ["previous_checkpoint_id"],
+            ["checkpoint.id"],
+            name=op.f("fk__checkpoint__previous_checkpoint_id__checkpoint"),
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk__checkpoint")),
+        sa.UniqueConstraint(
+            "group_id",
+            "document_id",
+            "previous_checkpoint_id",
+            name="uq__checkpoint__group_id__document_id__previous_checkpoint_id",
+            postgresql_nulls_not_distinct=True,
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("checkpoint")

h/models/__init__.py

@@ -25,6 +25,7 @@
 from h.models.auth_ticket import AuthTicket
 from h.models.authz_code import AuthzCode
 from h.models.blocklist import Blocklist
+from h.models.checkpoint import Checkpoint
 from h.models.document import Document, DocumentMeta, DocumentURI
 from h.models.feature import Feature
 from h.models.feature_cohort import FeatureCohort, FeatureCohortUser
@@ -54,6 +55,7 @@
     "AuthTicket",
     "AuthzCode",
     "Blocklist",
+    "Checkpoint",
     "Document",
     "DocumentMeta",
     "DocumentURI",

h/models/checkpoint.py

@@ -0,0 +1,59 @@
+from datetime import datetime
+
+import sqlalchemy as sa
+from sqlalchemy.orm import Mapped, mapped_column
+
+from h.db import Base
+from h.db.mixins import Timestamps
+from h.models import helpers
+
+
+class Checkpoint(Base, Timestamps):
+    """A hide/reveal checkpoint, synced from the LMS so h can authorize annotation visibility.
+
+    Checkpoints form a per-(group, document) linked list: a checkpoint's start
+    date is derived from its predecessor's reveal_date, and the first one
+    (previous_checkpoint_id NULL) starts at the assignment creation date.
+    Annotations stay hidden until their checkpoint's reveal_date passes.
+    """
+
+    __tablename__ = "checkpoint"
+
+    id: Mapped[int] = mapped_column(sa.Integer, autoincrement=True, primary_key=True)
+
+    group_id: Mapped[int] = mapped_column(
+        sa.Integer, sa.ForeignKey("group.id", ondelete="CASCADE"), nullable=False
+    )
+    group = sa.orm.relationship("Group")
+
+    document_id: Mapped[int] = mapped_column(
+        sa.Integer, sa.ForeignKey("document.id", ondelete="CASCADE"), nullable=False
+    )
+    document = sa.orm.relationship("Document")
+
+    previous_checkpoint_id: Mapped[int | None] = mapped_column(
+        sa.Integer, sa.ForeignKey("checkpoint.id", ondelete="CASCADE"), nullable=True
+    )
+    previous_checkpoint = sa.orm.relationship(
+        "Checkpoint", remote_side=[id], uselist=False
+    )
+
+    reveal_date: Mapped[datetime | None] = mapped_column(sa.DateTime, nullable=True)
+    """When the instructor reveals this checkpoint; NULL until revealed."""
+
+    __table_args__ = (
+        # NULLS NOT DISTINCT (PG15+) so the NULL-previous root is unique too:
+        # at most one first checkpoint per (group, uri).
+        sa.UniqueConstraint(
+            "group_id",
+            "document_id",
+            "previous_checkpoint_id",
+            name="uq__checkpoint__group_id__document_id__previous_checkpoint_id",
+            postgresql_nulls_not_distinct=True,
+        ),
+    )
+
+    def __repr__(self) -> str:
+        return helpers.repr_(
+            self, ["id", "group_id", "previous_checkpoint_id", "reveal_date"]
+        )

tests/common/factories/__init__.py

@@ -9,6 +9,7 @@
 from tests.common.factories.auth_ticket import AuthTicket
 from tests.common.factories.authz_code import AuthzCode
 from tests.common.factories.base import set_session
+from tests.common.factories.checkpoint import Checkpoint
 from tests.common.factories.document import Document, DocumentMeta, DocumentURI
 from tests.common.factories.feature import Feature
 from tests.common.factories.feature_cohort import FeatureCohort

tests/common/factories/checkpoint.py

@@ -0,0 +1,16 @@
+import factory
+
+from h import models
+
+from .base import ModelFactory
+from .document import Document
+from .group import Group
+
+
+class Checkpoint(ModelFactory):
+    class Meta:
+        model = models.Checkpoint
+        sqlalchemy_session_persistence = "flush"
+
+    group = factory.SubFactory(Group)
+    document = factory.SubFactory(Document)

tests/unit/h/models/checkpoint_test.py

@@ -0,0 +1,10 @@
+class TestCheckpoint:
+    def test___repr__(self, factories):
+        checkpoint = factories.Checkpoint()
+
+        assert repr(checkpoint) == (
+            f"Checkpoint(id={checkpoint.id!r}, "
+            f"group_id={checkpoint.group_id!r}, "
+            f"previous_checkpoint_id={checkpoint.previous_checkpoint_id!r}, "
+            f"reveal_date={checkpoint.reveal_date!r})"
+        )