[HnR:4] Bulk checkpoint upsert API + reveal response fields

h/routes.py

@@ -159,6 +159,14 @@ def includeme(config):  # noqa: PLR0915
     config.add_route(
         "api.bulk.lms.annotations", "/api/bulk/lms/annotations", request_method="POST"
     )
+    config.add_route(
+        "api.bulk.checkpoint", "/api/bulk/checkpoint", request_method="POST"
+    )
+    config.add_route(
+        "api.bulk.checkpoint.reveal",
+        "/api/bulk/checkpoint/reveal",
+        request_method="POST",
+    )
 
     config.add_route("api.groups", "/api/groups", factory="h.traversal.GroupRoot")
     config.add_route(

h/security/policy/_auth_client.py

@@ -39,6 +39,8 @@ class AuthClientPolicy:
         ("api.bulk.annotation", "POST"),
         ("api.bulk.group", "POST"),
         ("api.bulk.lms.annotations", "POST"),
+        ("api.bulk.checkpoint", "POST"),
+        ("api.bulk.checkpoint.reveal", "POST"),
     ]
 
     @classmethod

h/services/checkpoint.py

@@ -1,17 +1,21 @@
 from dataclasses import dataclass
-from datetime import datetime
+from datetime import UTC, datetime
+from typing import ClassVar
 
-from sqlalchemy import or_, select
+from sqlalchemy import func, or_, select
+from sqlalchemy.dialects.postgresql import insert
 
 from h.models import (
     Annotation,
     Checkpoint,
     Document,
     DocumentURI,
+    Group,
     GroupMembership,
     User,
 )
 from h.models.group import LMSRole
+from h.schemas import ValidationError
 
 
 @dataclass
@@ -126,6 +130,153 @@ def hides_annotation(self, user: User | None, annotation: Annotation) -> bool:
 
         return False
 
+    _ROLE_MAP: ClassVar[dict[str, LMSRole]] = {
+        "instructor": LMSRole.LMS_INSTRUCTOR,
+        "student": LMSRole.LMS_STUDENT,
+    }
+
+    def set_user_role(
+        self,
+        authority: str,
+        username: str,
+        role: str,
+        group_authority_provided_ids: list[str],
+    ) -> None:
+        """Set the lms_role for a user in the given groups."""
+        lms_role = self._ROLE_MAP.get(role)
+        if not lms_role:
+            return
+
+        user = User.get_by_username(self.db, username, authority)
+        if not user:
+            return
+
+        memberships = self.db.scalars(
+            select(GroupMembership)
+            .join(Group, Group.id == GroupMembership.group_id)
+            .where(GroupMembership.user_id == user.id)
+            .where(Group.authority == authority)
+            .where(Group.authority_provided_id.in_(group_authority_provided_ids))
+        ).all()
+
+        for membership in memberships:
+            membership.lms_role = lms_role.value
+
+    def upsert_checkpoint(
+        self,
+        authority: str,
+        group_authority_provided_id: str,
+        document_uri: str,
+        reveal_date: str | None = None,
+    ) -> Checkpoint | None:
+        """
+        Upsert a checkpoint for a (group, document) pair.
+
+        Resolves the group by authority + authority_provided_id and the
+        document by URI. Creates the document if it doesn't exist yet.
+
+        Returns the upserted Checkpoint, or None if the group or document
+        could not be resolved.
+        """
+        group = self.db.scalar(
+            select(Group).where(
+                Group.authority == authority,
+                Group.authority_provided_id == group_authority_provided_id,
+            )
+        )
+        if not group:
+            return None
+
+        document = Document.find_or_create_by_uris(
+            self.db, claimant_uri=document_uri, uris=[]
+        ).first()
+        if not document:
+            return None
+
+        parsed_reveal_date = None
+        if reveal_date:
+            try:
+                parsed_reveal_date = datetime.fromisoformat(reveal_date)
+            except ValueError as err:
+                msg = f"Invalid reveal_date: {reveal_date!r}"
+                raise ValidationError(msg) from err
+            # Store naive UTC to match the column and the utcnow() comparisons.
+            if parsed_reveal_date.tzinfo is not None:
+                parsed_reveal_date = parsed_reveal_date.astimezone(UTC).replace(
+                    tzinfo=None
+                )
+
+        stmt = (
+            insert(Checkpoint)
+            .values(
+                group_id=group.id,
+                document_id=document.id,
+                previous_checkpoint_id=None,
+                reveal_date=parsed_reveal_date,
+            )
+            # If a checkpoint already exists for this (group, document), update the
+            # reveal_date. coalesce keeps the existing date if the new one is NULL.
+            .on_conflict_do_update(
+                constraint="uq__checkpoint__group_id__document_id__previous_checkpoint_id",
+                set_={
+                    "reveal_date": func.coalesce(
+                        parsed_reveal_date, Checkpoint.reveal_date
+                    )
+                },
+            )
+            .returning(Checkpoint.id)
+        )
+        result = self.db.execute(stmt)
+        self.db.flush()
+
+        checkpoint_id = result.scalar()
+        return self.db.get(Checkpoint, checkpoint_id)
+
+    def reveal_checkpoints(
+        self,
+        authority: str,
+        group_authority_provided_id: str,
+        document_uri: str,
+    ) -> Checkpoint | None:
+        """
+        Reveal a checkpoint by setting its reveal_date to now.
+
+        Returns the updated Checkpoint, or None if not found.
+        """
+        group = self.db.scalar(
+            select(Group).where(
+                Group.authority == authority,
+                Group.authority_provided_id == group_authority_provided_id,
+            )
+        )
+        if not group:
+            return None
+
+        document_ids = [
+            doc.id for doc in Document.find_by_uris(self.db, [document_uri])
+        ]
+        if not document_ids:
+            return None
+
+        checkpoint = self.db.scalar(
+            select(Checkpoint)
+            .where(Checkpoint.group_id == group.id)
+            .where(Checkpoint.document_id.in_(document_ids))
+            .where(
+                or_(
+                    Checkpoint.reveal_date.is_(None),
+                    Checkpoint.reveal_date > datetime.utcnow(),  # noqa: DTZ003
+                )
+            )
+            .limit(1)
+        )
+        if not checkpoint:
+            return None
+
+        checkpoint.reveal_date = datetime.utcnow()  # noqa: DTZ003
+        self.db.flush()
+        return checkpoint
+
     def _hidden_scope(self, user: User, checkpoint: Checkpoint) -> HiddenScope:
         group_pubid = checkpoint.group.pubid
 

h/views/api/bulk/checkpoint.py

@@ -0,0 +1,116 @@
+import json
+from datetime import datetime
+
+from importlib_resources import files
+from pyramid.response import Response
+
+from h.schemas.base import JSONSchema
+from h.security import Permission
+from h.services.checkpoint import CheckpointService
+from h.views.api.config import api_config
+
+
+def _is_revealed(checkpoint):
+    return (
+        checkpoint is not None
+        and checkpoint.reveal_date is not None
+        and checkpoint.reveal_date <= datetime.utcnow()  # noqa: DTZ003
+    )
+
+
+def _reveal_date_isoformat(checkpoint):
+    if checkpoint and checkpoint.reveal_date:
+        return checkpoint.reveal_date.isoformat()
+    return None
+
+
+class BulkCheckpointSchema(JSONSchema):
+    _SCHEMA_FILE = files("h.views.api.bulk") / "checkpoint_schema.json"
+    schema_version = 7
+    schema = json.loads(_SCHEMA_FILE.read_text(encoding="utf-8"))
+
+
+class BulkCheckpointRevealSchema(JSONSchema):
+    _SCHEMA_FILE = files("h.views.api.bulk") / "checkpoint_reveal_schema.json"
+    schema_version = 7
+    schema = json.loads(_SCHEMA_FILE.read_text(encoding="utf-8"))
+
+
+@api_config(
+    versions=["v1", "v2"],
+    route_name="api.bulk.checkpoint",
+    request_method="POST",
+    link_name="bulk.checkpoint",
+    description="Upsert checkpoints",
+    permission=Permission.API.BULK_ACTION,
+)
+def upsert_checkpoints(request):
+    data = BulkCheckpointSchema().validate(request.json)
+    authority = request.identity.auth_client.authority
+
+    checkpoint_service = request.find_service(CheckpointService)
+
+    group_authority_provided_ids = [
+        item["group_authority_provided_id"] for item in data["checkpoints"]
+    ]
+
+    if user_data := data.get("user"):
+        checkpoint_service.set_user_role(
+            authority=authority,
+            username=user_data["username"],
+            role=user_data["role"],
+            group_authority_provided_ids=group_authority_provided_ids,
+        )
+
+    results = []
+    for item in data["checkpoints"]:
+        checkpoint = checkpoint_service.upsert_checkpoint(
+            authority=authority,
+            group_authority_provided_id=item["group_authority_provided_id"],
+            document_uri=item["document_uri"],
+            reveal_date=item.get("reveal_date"),
+        )
+        results.append(
+            {
+                "group_authority_provided_id": item["group_authority_provided_id"],
+                "document_uri": item["document_uri"],
+                "created": checkpoint is not None,
+                "revealed": _is_revealed(checkpoint),
+                "reveal_date": _reveal_date_isoformat(checkpoint),
+            }
+        )
+
+    return Response(json=results, status=200)
+
+
+@api_config(
+    versions=["v1", "v2"],
+    route_name="api.bulk.checkpoint.reveal",
+    request_method="POST",
+    link_name="bulk.checkpoint.reveal",
+    description="Reveal checkpoints, making annotations visible immediately",
+    permission=Permission.API.BULK_ACTION,
+)
+def reveal_checkpoints(request):
+    data = BulkCheckpointRevealSchema().validate(request.json)
+    authority = request.identity.auth_client.authority
+
+    checkpoint_service = request.find_service(CheckpointService)
+
+    results = []
+    for item in data["checkpoints"]:
+        checkpoint = checkpoint_service.reveal_checkpoints(
+            authority=authority,
+            group_authority_provided_id=item["group_authority_provided_id"],
+            document_uri=item["document_uri"],
+        )
+        results.append(
+            {
+                "group_authority_provided_id": item["group_authority_provided_id"],
+                "document_uri": item["document_uri"],
+                "revealed": checkpoint is not None,
+                "reveal_date": _reveal_date_isoformat(checkpoint),
+            }
+        )
+
+    return Response(json=results, status=200)

h/views/api/bulk/checkpoint_reveal_schema.json

@@ -0,0 +1,29 @@
+{
+  "$schema": "https://json-schema.org/draft-07/schema",
+  "type": "object",
+  "title": "Bulk checkpoint reveal",
+  "properties": {
+    "authority": {
+      "type": "string"
+    },
+    "checkpoints": {
+      "type": "array",
+      "minItems": 1,
+      "items": {
+        "type": "object",
+        "properties": {
+          "group_authority_provided_id": {
+            "type": "string"
+          },
+          "document_uri": {
+            "type": "string"
+          }
+        },
+        "required": ["group_authority_provided_id", "document_uri"],
+        "additionalProperties": false
+      }
+    }
+  },
+  "required": ["authority", "checkpoints"],
+  "additionalProperties": false
+}

h/views/api/bulk/checkpoint_schema.json

@@ -0,0 +1,46 @@
+{
+  "$schema": "https://json-schema.org/draft-07/schema",
+  "type": "object",
+  "title": "Bulk checkpoint upsert",
+  "properties": {
+    "authority": {
+      "type": "string"
+    },
+    "checkpoints": {
+      "type": "array",
+      "minItems": 1,
+      "items": {
+        "type": "object",
+        "properties": {
+          "group_authority_provided_id": {
+            "type": "string"
+          },
+          "document_uri": {
+            "type": "string"
+          },
+          "reveal_date": {
+            "type": ["string", "null"]
+          }
+        },
+        "required": ["group_authority_provided_id", "document_uri"],
+        "additionalProperties": false
+      }
+    },
+    "user": {
+      "type": "object",
+      "properties": {
+        "username": {
+          "type": "string"
+        },
+        "role": {
+          "type": "string",
+          "enum": ["instructor", "student"]
+        }
+      },
+      "required": ["username", "role"],
+      "additionalProperties": false
+    }
+  },
+  "required": ["authority", "checkpoints"],
+  "additionalProperties": false
+}