[HnR:2] Hide & Reveal: assignment config, checkpoint sync, and reveal API (backend)

lms/product/canvas/_plugin/misc.py

@@ -59,14 +59,17 @@ def get_assignment_configuration(
             group_set_id=request.params.get("group_set"),
         )
 
-        if auto_grading_config := self.get_deep_linked_assignment_configuration(
-            request
-        ).get("auto_grading_config"):
+        deep_linked_config = self.get_deep_linked_assignment_configuration(request)
+
+        if auto_grading_config := deep_linked_config.get("auto_grading_config"):
             # Auto grading is a complex structure, deserialize it beforehand
             assignment_config["auto_grading_config"] = cast(
                 "AutoGradingConfig", json.loads(auto_grading_config)
             )
 
+        if deep_linked_config.get("checkpoint_enabled") in ("true", True):
+            assignment_config["checkpoint_enabled"] = True
+
         return assignment_config
 
     @lru_cache(1)  # noqa: B019
@@ -150,6 +153,7 @@ def get_deep_linked_assignment_configuration(self, request) -> dict:
         possible_parameters = [
             "group_set",
             "auto_grading_config",
+            "checkpoint_enabled",
             # VS, legacy method
             "vitalsource_book",
             "book_id",

lms/product/plugin/misc.py

@@ -15,6 +15,7 @@ class AssignmentConfig(TypedDict):
     document_url: str | None
     group_set_id: str | None
     auto_grading_config: NotRequired[AutoGradingConfig | None]
+    checkpoint_enabled: NotRequired[bool]
 
 
 class DeepLinkingPromptForGradableMixin:
@@ -125,6 +126,7 @@ def get_deep_linked_assignment_configuration(self, request) -> dict:
             "group_set",
             "deep_linking_uuid",
             "auto_grading_config",
+            "checkpoint_enabled",
         ]
 
         for param in possible_parameters:
@@ -148,6 +150,9 @@ def _assignment_config_from_assignment(assignment: Assignment) -> AssignmentConf
         if auto_grading_config := assignment.auto_grading_config:
             config["auto_grading_config"] = auto_grading_config.asdict()
 
+        if assignment.checkpoint_enabled:
+            config["checkpoint_enabled"] = True
+
         return config
 
     @staticmethod
@@ -163,4 +168,7 @@ def _assignment_config_from_deep_linked_config(
         if auto_grading_config := deep_linked_config.get("auto_grading_config"):
             config["auto_grading_config"] = json.loads(auto_grading_config)
 
+        if deep_linked_config.get("checkpoint_enabled") in ("true", True):
+            config["checkpoint_enabled"] = True
+
         return config

lms/resources/_js_config/__init__.py

@@ -1,6 +1,6 @@
 import functools
 import re
-from datetime import timedelta
+from datetime import UTC, timedelta
 from enum import Enum, StrEnum
 from typing import Any
 from urllib.parse import urlparse
@@ -471,6 +471,43 @@ def enable_instructor_dashboard_entry_point(self, assignment):
         }
         self._config["hypothesisClient"] = self._hypothesis_client
 
+    def enable_toolbar_checkpoint(
+        self, assignment, *, h_revealed=False, h_reveal_date=None
+    ):
+        toolbar_config = self._config.get("instructorToolbar", {})
+
+        due_date_iso = (
+            assignment.due_date.replace(tzinfo=UTC).isoformat()
+            if assignment.due_date
+            else None
+        )
+
+        toolbar_config["courseCheckpointConfig"] = {
+            "revealed": h_revealed,
+            "revealDate": h_reveal_date,
+            "revealUrl": self._request.route_url(
+                "api.checkpoint.reveal", assignment_id=assignment.id
+            ),
+        }
+        toolbar_config["assignmentDueDate"] = due_date_iso
+        toolbar_config["assignmentCheckpointEnabled"] = True
+        self._config["instructorToolbar"] = toolbar_config
+
+    def enable_student_checkpoint(self, assignment, *, h_revealed=False):
+        due_date_iso = (
+            assignment.due_date.replace(tzinfo=UTC).isoformat()
+            if assignment.due_date
+            else None
+        )
+
+        self._config["studentToolbar"] = {
+            "courseCheckpointConfig": {
+                "revealed": h_revealed,
+            },
+            "assignmentDueDate": due_date_iso,
+            "assignmentCheckpointEnabled": True,
+        }
+
     def enable_toolbar_editing(self):
         toolbar_config = self._get_toolbar_config()
 

lms/routes.py

@@ -45,6 +45,11 @@ def includeme(config):  # noqa: PLR0915
     )
 
     config.add_route("api.sync", "/api/sync", request_method="POST")
+    config.add_route(
+        "api.checkpoint.reveal",
+        "/api/assignments/{assignment_id}/checkpoint/reveal",
+        request_method="POST",
+    )
     config.add_route(
         "api.courses.group_sets.list", "/api/courses/{course_id}/group_sets"
     )

lms/services/assignment.py

@@ -62,6 +62,7 @@ def update_assignment(  # noqa: PLR0913
         group_set_id,
         course: Course,
         auto_grading_config: dict | None = None,
+        checkpoint_enabled: bool = False,  # noqa: FBT001, FBT002
     ):
         """Update an existing assignment."""
         if self._misc_plugin.is_speed_grader_launch(request):
@@ -96,6 +97,7 @@ def update_assignment(  # noqa: PLR0913
 
         assignment.course_id = course.id
         self._update_auto_grading_config(assignment, auto_grading_config)
+        self._update_checkpoint(assignment, checkpoint_enabled)
 
         return assignment
 
@@ -151,6 +153,7 @@ def get_assignment_for_launch(self, request, course: Course) -> Assignment | Non
         document_url = assignment_config.get("document_url")
         group_set_id = assignment_config.get("group_set_id")
         auto_grading_config = assignment_config.get("auto_grading_config")
+        checkpoint_enabled = assignment_config.get("checkpoint_enabled", False)
 
         if not document_url:
             # We can't find a document_url, we shouldn't try to create an
@@ -181,7 +184,13 @@ def get_assignment_for_launch(self, request, course: Course) -> Assignment | Non
         # It often will be the same one while launching the assignment again but
         # it might for example be an updated deep linked URL or similar.
         return self.update_assignment(
-            request, assignment, document_url, group_set_id, course, auto_grading_config
+            request,
+            assignment,
+            document_url,
+            group_set_id,
+            course,
+            auto_grading_config,
+            checkpoint_enabled=checkpoint_enabled,
         )
 
     def upsert_assignment_membership(
@@ -373,6 +382,19 @@ def get_assignment_sections(self, assignment) -> Sequence[Grouping]:
             .order_by(Grouping.lms_name.asc())
         ).all()
 
+    def _update_checkpoint(
+        self,
+        assignment: Assignment,
+        checkpoint_enabled: bool,  # noqa: FBT001
+    ) -> None:
+        """Mark an assignment as checkpoint_enabled.
+
+        Checkpoints can only be enabled at creation time -- once enabled
+        they are never disabled by an edit.
+        """
+        if checkpoint_enabled and not assignment.checkpoint_enabled:
+            assignment.checkpoint_enabled = True
+
     def _update_auto_grading_config(
         self, assignment: Assignment, auto_grading_config: dict | None
     ) -> None:

lms/services/h_api.py

@@ -217,6 +217,72 @@ def get_annotation_counts(
         )
         return response.json()
 
+    def sync_checkpoints(
+        self,
+        authority: str,
+        checkpoints: list[dict],
+        user: dict | None = None,
+    ) -> list[dict] | None:
+        """Sync checkpoint data to h via the bulk checkpoint endpoint.
+
+        :param authority: The h authority
+        :param checkpoints: List of dicts with group_authority_provided_id,
+            document_uri, and optionally reveal_date
+        :param user: Optional dict with 'username' and 'role' to set
+            their lms_role in the group memberships
+        :return: List of checkpoint results from h, each containing
+            revealed and reveal_date fields, or None if no checkpoints
+        """
+        if not checkpoints:
+            return None
+
+        payload: dict = {
+            "authority": authority,
+            "checkpoints": checkpoints,
+        }
+        if user:
+            payload["user"] = user
+
+        response = self._api_request(
+            "POST",
+            path="bulk/checkpoint",
+            body=json.dumps(payload),
+            headers={
+                "Content-Type": "application/json",
+            },
+        )
+        return response.json()
+
+    def reveal_checkpoints(
+        self,
+        authority: str,
+        checkpoints: list[dict],
+    ) -> list[dict] | None:
+        """Reveal checkpoints in h, making annotations visible immediately.
+
+        :param authority: The h authority
+        :param checkpoints: List of dicts with group_authority_provided_id
+            and document_uri
+        :return: List of reveal results from h, or None if no checkpoints
+        """
+        if not checkpoints:
+            return None
+
+        payload = {
+            "authority": authority,
+            "checkpoints": checkpoints,
+        }
+
+        response = self._api_request(
+            "POST",
+            path="bulk/checkpoint/reveal",
+            body=json.dumps(payload),
+            headers={
+                "Content-Type": "application/json",
+            },
+        )
+        return response.json()
+
     def _api_request(self, method, path, body=None, headers=None, stream=False):  # noqa: FBT002
         """
         Send any kind of HTTP request to the h API and return the response.

lms/services/lti_h.py

@@ -1,9 +1,33 @@
 from h_api.bulk_api import CommandBuilder
 
-from lms.models import Grouping
+from lms.models import Assignment, Grouping
 from lms.services import HAPI
 
 
+def checkpoint_sync_data(assignment: Assignment | None, lti_user) -> dict | None:
+    """Build the checkpoint payload to sync to h for a Hide & Reveal assignment.
+
+    Returns None when the assignment is missing or doesn't have checkpoint
+    enabled, so callers can pass the result straight through to
+    `LTIHService.sync(..., checkpoint_data=...)`.
+
+    reveal_date is not sent — h is the source of truth for the reveal state.
+    h's upsert uses coalesce to preserve an existing reveal_date when NULL
+    is sent.
+    """
+    if not (assignment and assignment.checkpoint_enabled):
+        return None
+
+    role = "instructor" if lti_user.is_instructor else "student"
+    return {
+        "document_uri": assignment.document_url,
+        "user": {
+            "username": lti_user.h_user.username,
+            "role": role,
+        },
+    }
+
+
 class LTIHService:
     """
     Copy LTI users and courses to h users and groups.
@@ -25,7 +49,12 @@ def __init__(self, _context, request) -> None:
         self._h_api: HAPI = request.find_service(HAPI)
         self._group_info_service = request.find_service(name="group_info")
 
-    def sync(self, groupings: list[Grouping], group_info_params: dict):
+    def sync(
+        self,
+        groupings: list[Grouping],
+        group_info_params: dict,
+        checkpoint_data: dict | None = None,
+    ) -> list[dict] | None:
         """
         Sync standard data to h for an LTI launch with the provided groups.
 
@@ -34,6 +63,9 @@ def sync(self, groupings: list[Grouping], group_info_params: dict):
 
         :param groupings: groupings to sync to H
         :param group_info_params: params to add for each in `GroupInfo`
+        :param checkpoint_data: optional dict with document_uri and reveal_date
+            to sync a checkpoint for each grouping
+        :return: checkpoint results from h, or None
 
         :raise HTTPInternalServerError: if we can't sync to h for any reason
         :raise ApplicationInstanceNotFound: if
@@ -47,6 +79,10 @@ def sync(self, groupings: list[Grouping], group_info_params: dict):
                 grouping=grouping, params=group_info_params
             )
 
+        if checkpoint_data:
+            return self._sync_checkpoints(groupings, checkpoint_data)
+        return None
+
     def _yield_commands(self, groupings):
         # Note! - Syncing a user to `h` currently has an implication for
         # reporting and so billing and will as long as our billing metric is
@@ -77,6 +113,23 @@ def _user_upsert(self, h_user, ref="user_0"):
             ref,
         )
 
+    def _sync_checkpoints(
+        self, groupings: list[Grouping], checkpoint_data: dict
+    ) -> list[dict] | None:
+        checkpoints = [
+            {
+                "group_authority_provided_id": grouping.authority_provided_id,
+                "document_uri": checkpoint_data["document_uri"],
+            }
+            for grouping in groupings
+        ]
+
+        return self._h_api.sync_checkpoints(
+            authority=self._authority,
+            checkpoints=checkpoints,
+            user=checkpoint_data.get("user"),
+        )
+
     def _group_upsert(self, grouping, ref):
         return CommandBuilder.group.upsert(
             {

lms/validation/_lti_launch_params.py

@@ -185,6 +185,12 @@ class ConfigureAssignmentSchema(_CommonLTILaunchSchema):
     auto_grading_config = fields.Nested(
         AutoGradingConfigSchema, required=False, allow_none=True
     )
+    checkpoint_enabled = fields.Bool(
+        required=False,
+        load_default=False,
+        truthy={"true", "1"},
+        falsy={"false", "0", ""},
+    )
 
     @pre_load
     def _load_auto_grading_config(self, data, **_kwargs):