The Scenario
An online Python bootcamp needs live progress dashboards, real-time quiz grading, and instructor alerts — all without a JS build chain. Students should see their module completion update the moment they finish a lesson. Instructors need a live table reflecting every student's progress as it happens.
Four components cover the entire LMS surface:
- CourseProgressComponent — tracks module completion per student and emits a WebSocket broadcast whenever a module is marked finished, so instructors see the update immediately.
- QuizComponent — handles quiz submission with immediate server-side grading and per-question feedback, rate-limited to prevent exam abuse.
- InstructorDashboardComponent — a live table of all enrolled students' progress; subscribes to the course WebSocket channel and patches individual rows in-memory when any student advances.
- GradeNotificationComponent — push notifications delivered directly to a student's browser when an instructor posts a grade, using a per-student WebSocket channel.
Architecture at a Glance
Key principle: the server owns all state. The browser is a dumb renderer that sends events and receives fresh HTML fragments. No synchronisation logic, no client state machine, no Redux.
CourseProgressComponent
Tracks which modules a student has completed, calculates overall percentage, and broadcasts a WebSocket message to the instructor channel whenever a module is finished.
The Component
from component_framework.core import Component, registry, ws_manager
from component_framework.core.permissions import IsAuthenticated
from .models import Progress
@registry.register("course_progress")
class CourseProgressComponent(Component):
"""
Renders per-student module progress and broadcasts completion events
to the course instructor channel via WebSocket.
params:
course_id (int): The course being tracked.
"""
template_name = "components/course_progress.html"
permission_classes = [IsAuthenticated]
def mount(self):
student = self.request.user
self.state["course_id"] = self.params["course_id"]
self.state["modules"] = self._load_modules(student)
self.state["completed"] = self._load_completed(student)
self.state["pct"] = self._calc_pct()
async def on_complete_module(self, module_id: int):
"""Mark a module complete and notify all instructors."""
if module_id in self.state["completed"]:
return # idempotent — already done
await Progress.objects.acreate(
user=self.request.user,
module_id=module_id,
)
self.state["completed"].append(module_id)
self.state["pct"] = self._calc_pct()
# Push live update to every instructor watching this course
await ws_manager.broadcast(
f"instructor-{self.state['course_id']}",
{
"event": "module_completed",
"student": self.request.user.username,
"module": module_id,
"pct": self.state["pct"],
},
)
def _load_modules(self, student) -> list:
from .models import Module
return list(
Module.objects
.filter(course_id=self.params["course_id"])
.values_list("id", flat=True)
)
def _load_completed(self, student) -> list:
return list(
Progress.objects
.filter(user=student, module__course_id=self.params["course_id"])
.values_list("module_id", flat=True)
)
def _calc_pct(self) -> int:
total = len(self.state["modules"])
return int(len(self.state["completed"]) / total * 100) if total else 0
The Template
<div id="course-progress-{{ component_id }}"
hx-target="this"
hx-swap="outerHTML">
<div class="progress-header">
<span class="progress-pct">{{ state.pct }}% complete</span>
<span class="progress-count">
{{ state.completed|length }} / {{ state.modules|length }} modules
</span>
</div>
{# Progress bar #}
<div class="progress-bar">
<div class="progress-bar__fill"
style="width: {{ state.pct }}%"></div>
</div>
{# Module list — click to mark complete #}
<ul class="module-list">
{% for mod_id in state.modules %}
<li class="module-item {% if mod_id in state.completed %}done{% endif %}">
{% if mod_id not in state.completed %}
<button
hx-post="{% url 'course_progress_component' %}"
hx-vals='{"event": "complete_module", "module_id": {{ mod_id }}}'>
Mark complete
</button>
{% else %}
<span class="checkmark">✓</span>
{% endif %}
Module {{ mod_id }}
</li>
{% endfor %}
</ul>
</div>
on_complete_module() writes a Progress row and immediately broadcasts to the instructor-{course_id} channel. Every connected instructor dashboard patches its in-memory student row without a page reload or a polling interval.
QuizComponent
Handles quiz submission with Pydantic validation, immediate server-side grading, and per-question feedback. Backed by FormComponent so the schema is the single source of truth for both validation and the rendered form.
The Component
from pydantic import BaseModel
from component_framework.core.form import FormComponent
from component_framework.core.permissions import IsAuthenticated
from .models import Quiz, Submission
from .permissions import IsEnrolled
class QuizAnswersSchema(BaseModel):
answers: dict[int, str] # question_id → selected answer
@registry.register("quiz")
class QuizComponent(FormComponent):
"""
Server-side quiz grading with per-question feedback.
permission_classes enforces both authentication and course enrolment.
rate_limit prevents students from brute-forcing answers.
"""
schema = QuizAnswersSchema
template_name = "components/quiz.html"
permission_classes = [IsAuthenticated, IsEnrolled]
rate_limit = "3/hour"
def mount(self, quiz_id: int):
self.state["quiz"] = Quiz.objects.get(pk=quiz_id)
self.state["submitted"] = False
self.state["score"] = None
self.state["feedback"] = {}
def on_submit(self):
"""Grade the submission and persist it."""
answers = self.validated_data["answers"]
score, feedback = self.grade(answers)
Submission.objects.create(
user=self.request.user,
quiz=self.state["quiz"],
score=score,
)
self.state["submitted"] = True
self.state["score"] = score
self.state["feedback"] = feedback
def grade(self, answers: dict) -> tuple[int, dict]:
"""
Compare submitted answers against the answer key.
Returns (score_pct, {question_id: 'correct'|'incorrect'}).
"""
quiz = self.state["quiz"]
correct = 0
feedback = {}
for question in quiz.questions.all():
given = answers.get(question.pk, "")
ok = given == question.correct_answer
feedback[question.pk] = "correct" if ok else "incorrect"
if ok:
correct += 1
total = quiz.questions.count()
score = int(correct / total * 100) if total else 0
return score, feedback
Rate Limiting Quiz Submissions
Retry-After header. The limit applies per authenticated user, so one student cannot affect another's quota. No API gateway, no custom middleware — one attribute.
FormComponent runs Pydantic validation before calling on_submit(). If the payload fails the QuizAnswersSchema schema — missing fields, wrong types — the component returns validation errors and never reaches the grading logic. self.validated_data is always type-safe.
InstructorDashboardComponent
Mounts once and shows a table of every enrolled student's current completion percentage. It subscribes to the course WebSocket channel and patches individual rows in-memory when any student advances — no polling, no full-page reload.
Component with WebSocket Subscription
from django.db.models import Count
from component_framework.core.permissions import IsAuthenticated
from .permissions import IsInstructor
@registry.register("instructor_dashboard")
class InstructorDashboardComponent(Component):
"""
Live instructor view of all student progress for a course.
WebSocket channel: instructor-{course_id}
Patches individual student rows on module_completed events.
"""
template_name = "components/instructor_dashboard.html"
permission_classes = [IsAuthenticated, IsInstructor]
cache_key_prefix = "instructor_dash"
cache_ttl = 30 # seconds — event requests bypass automatically
def mount(self, course_id: int):
self.state["course_id"] = course_id
self.state["students"] = self._load_all_progress(course_id)
async def on_ws_message(self, message: dict):
"""
Handle WebSocket push from CourseProgressComponent.
Patches one student row — no DB query, no full reload.
"""
if message.get("event") != "module_completed":
return
username = message["student"]
for student in self.state["students"]:
if student["username"] == username:
student["pct"] = message["pct"]
break
def _load_all_progress(self, course_id: int) -> list:
from .models import Progress, Module
total_modules = Module.objects.filter(course_id=course_id).count()
rows = (
Progress.objects
.filter(module__course_id=course_id)
.values("user__username")
.annotate(completed=Count("module"))
.order_by("user__username")
)
return [
{
"username": r["user__username"],
"completed": r["completed"],
"pct": int(r["completed"] / total_modules * 100)
if total_modules else 0,
}
for r in rows
]
ws://…/ws/instructor-{course_id}/. The Django Channels consumer joins the instructor-{course_id} group. When any student fires on_complete_module(), the broadcast lands in every connected instructor's consumer, which calls on_ws_message(), patches the in-memory state, and sends a fresh HTML fragment to the browser. The instructor sees the row update in under 50 ms.
Real-time Grade Notifications
When an instructor posts a score, a WebSocket broadcast delivers the notification directly to that student's browser — no polling, no refresh needed. The broadcast targets a per-student channel so other students are never affected.
from component_framework.adapters.django_model import DjangoModelComponent
@registry.register("grade_poster")
class GradePosterComponent(DjangoModelComponent):
"""
Instructor tool for posting a score against a quiz Submission.
Broadcasts the result directly to the student's WebSocket channel.
"""
model = Submission
state_fields = ["score", "feedback"]
permission_classes = [IsAuthenticated, IsInstructor]
async def on_post_grade(self, score: int, feedback: str):
"""Save the grade and push a notification to the student."""
self.state["score"] = score
self.state["feedback"] = feedback
self.update_instance_from_state()
await self.save_instance()
# Push directly to the student who submitted
await ws_manager.broadcast(
f"student-{self.instance.user_id}",
{
"event": "grade_posted",
"quiz": self.instance.quiz.title,
"score": score,
},
)
@registry.register("grade_notification")
class GradeNotificationComponent(Component):
"""
Student-side notification banner.
Subscribes to the student's personal channel and renders
incoming grade_posted events as dismissible alerts.
"""
template_name = "components/grade_notification.html"
permission_classes = [IsAuthenticated]
def mount(self):
self.state["notifications"] = []
async def on_ws_message(self, message: dict):
if message.get("event") == "grade_posted":
self.state["notifications"].insert(0, {
"quiz": message["quiz"],
"score": message["score"],
})
def on_dismiss(self, index: int):
try:
del self.state["notifications"][index]
except IndexError:
pass
ws_manager.broadcast("student-{user_id}", …) sends to exactly one student's channel group. Other students — even those in the same course — never receive the message. No filtering logic on the client side is needed.
Component Composition
The full student view is assembled from independent components using CompositeComponent. Each sub-component manages its own state; the composite wires them together and renders them in a single template.
from component_framework.core.composition import CompositeComponent
@registry.register("student_view")
class StudentView(CompositeComponent):
"""
Top-level student page — assembles progress tracker,
grade notifications, and the active quiz into one view.
"""
template_name = "components/student_view.html"
components = {
"progress": "course_progress",
"notifications": "grade_notification",
}
The composite template delegates rendering to each child using the {% live_component %} template tag:
{% load components %}
<div class="student-layout">
{# Grade notifications — WebSocket, no polling #}
{% live_component "grade_notification" %}
{# Module progress bar + checklist #}
{% live_component "course_progress" course_id=course.id %}
{# Active quiz — shown only when one is assigned #}
{% if current_quiz %}
<div id="quiz-area">
{% live_component "quiz" quiz_id=current_quiz.id %}
</div>
{% endif %}
</div>
QuizComponent for a timed variant or adding a VideoComponent slot requires no changes to the other components or the composite template — only a new registration in registry.
Testing
Because components are pure Python, the full LMS behaviour is testable without Django, HTMX, or a running server. dispatch_event() mounts the component, fires the event, and returns the updated instance — all in memory.
import pytest
from django.contrib.auth.models import AnonymousUser
from component_framework.testing import (
ComponentTestCase,
dispatch_event,
assert_state,
assert_permission_denied,
)
from component_framework.adapters.django_ratelimit import RateLimitExceeded
from lms.components import QuizComponent
class TestQuizComponent(ComponentTestCase):
def test_correct_answer_scores_100(self):
component = dispatch_event(
QuizComponent,
"submit",
{"answers": {1: "B", 2: "True"}},
quiz_id=1,
)
assert_state(component, score=100, submitted=True)
def test_wrong_answers_score_proportionally(self):
# Quiz 2 has 4 questions; answer only 2 correctly
component = dispatch_event(
QuizComponent,
"submit",
{"answers": {1: "A", 2: "B", 3: "X", 4: "X"}},
quiz_id=2,
)
assert_state(component, score=50)
def test_rate_limit_blocks_fourth_attempt(self):
for _ in range(3):
dispatch_event(QuizComponent, "submit", {"answers": {}}, quiz_id=1)
with pytest.raises(RateLimitExceeded):
dispatch_event(QuizComponent, "submit", {"answers": {}}, quiz_id=1)
def test_unauthenticated_cannot_submit(self):
assert_permission_denied(
QuizComponent,
"submit",
user=AnonymousUser(),
)
def test_feedback_labels_each_question(self):
component = dispatch_event(
QuizComponent,
"submit",
{"answers": {1: "B", 2: "False"}},
quiz_id=1,
)
assert component.state["feedback"][1] == "correct"
assert component.state["feedback"][2] == "incorrect"
from component_framework.testing import ComponentTestCase, dispatch_event
from lms.components import CourseProgressComponent
class TestCourseProgressComponent(ComponentTestCase):
def test_completion_pct_updates(self):
component = dispatch_event(
CourseProgressComponent,
"complete_module",
{"module_id": 1},
course_id=1,
)
assert 1 in component.state["completed"]
assert component.state["pct"] > 0
def test_completing_same_module_twice_is_idempotent(self):
component = dispatch_event(
CourseProgressComponent,
"complete_module",
{"module_id": 1},
course_id=1,
)
before = component.state["pct"]
component.on_complete_module(1) # second call, same module
assert component.state["pct"] == before
assert component.state["completed"].count(1) == 1
assert_state(), assert_permission_denied(), and dispatch_event() are included in component_framework.testing with no extra setup.
Summary
| Component | Lines of Python | WebSocket | Framework |
|---|---|---|---|
| CourseProgressComponent | ~40 | broadcast | Django |
| QuizComponent | ~35 | — | Django |
| InstructorDashboardComponent | ~45 | subscribe | Django |
| GradePosterComponent | ~25 | broadcast | Django |
| GradeNotificationComponent | ~20 | subscribe | Django |
| StudentView (composite) | ~10 | — | Django |