SafePath Training — Phase 1B: Course Management
Overview
This document covers the complete course management subsystem:
- Pydantic schemas for courses, lessons, quizzes
- Course service with CRUD, publishing, versioning
- API endpoints for course management
- Frontend course list and course creation wizard
Prerequisite: Phase 1A (database schema) must be complete.
Files to create:
tellus-ehs-hazcom-service/app/schemas/safepath/__init__.pytellus-ehs-hazcom-service/app/schemas/safepath/course.pytellus-ehs-hazcom-service/app/schemas/safepath/quiz.pytellus-ehs-hazcom-service/app/services/safepath/__init__.pytellus-ehs-hazcom-service/app/services/safepath/course_service.pytellus-ehs-hazcom-service/app/api/v1/safepath/__init__.pytellus-ehs-hazcom-service/app/api/v1/safepath/courses.pytellus-ehs-hazcom-ui/src/types/safepath.tstellus-ehs-hazcom-ui/src/services/safepath-api.tstellus-ehs-hazcom-ui/src/pages/safepath/courses/index.tsxtellus-ehs-hazcom-ui/src/pages/safepath/courses/create.tsx
Pydantic Schemas
Course Schemas
File: tellus-ehs-hazcom-service/app/schemas/safepath/course.py
"""SafePath Course Schemas
Pydantic schemas for course, lesson, and content management operations.
"""
from datetime import datetime
from typing import Optional, List
from uuid import UUID
from pydantic import BaseModel, Field, ConfigDict
# ============================================================================
# Course Category Schemas
# ============================================================================
class CourseCategoryResponse(BaseModel):
"""Response schema for course categories."""
model_config = ConfigDict(from_attributes=True)
category_id: UUID
name: str
description: Optional[str] = None
is_system: bool
sort_order: int
# ============================================================================
# Lesson Schemas
# ============================================================================
class LessonCreate(BaseModel):
"""Schema for creating a lesson within a course."""
title: str = Field(..., min_length=1, max_length=255)
lesson_type: str = Field(..., pattern="^(video|pdf|slides|text|external_link)$")
content: Optional[dict] = None
sort_order: int = Field(default=0, ge=0)
completion_threshold_percent: int = Field(default=80, ge=0, le=100)
locale: str = Field(default="en", pattern="^(en|es)$")
class LessonUpdate(BaseModel):
"""Schema for updating a lesson."""
title: Optional[str] = Field(None, min_length=1, max_length=255)
lesson_type: Optional[str] = Field(None, pattern="^(video|pdf|slides|text|external_link)$")
content: Optional[dict] = None
sort_order: Optional[int] = Field(None, ge=0)
completion_threshold_percent: Optional[int] = Field(None, ge=0, le=100)
class LessonAssetResponse(BaseModel):
"""Response schema for a lesson asset."""
model_config = ConfigDict(from_attributes=True)
asset_id: UUID
file_name: str
file_type: str
file_size_bytes: Optional[int] = None
download_url: Optional[str] = None # Presigned S3 URL, generated at response time
class LessonResponse(BaseModel):
"""Response schema for a lesson."""
model_config = ConfigDict(from_attributes=True)
lesson_id: UUID
title: str
lesson_type: str
content: Optional[dict] = None
sort_order: int
completion_threshold_percent: int
locale: str
assets: List[LessonAssetResponse] = []
created_at: datetime
# ============================================================================
# Course Schemas
# ============================================================================
class CourseCreate(BaseModel):
"""Schema for creating a new course."""
title: str = Field(..., min_length=1, max_length=255)
description: Optional[str] = None
category_id: Optional[UUID] = None
osha_standard_ref: Optional[str] = Field(None, max_length=50)
estimated_duration_minutes: Optional[int] = Field(None, ge=1)
passing_score_percent: int = Field(default=80, ge=0, le=100)
max_retakes: int = Field(default=3, ge=0)
locale: str = Field(default="en", pattern="^(en|es)$")
class CourseUpdate(BaseModel):
"""Schema for updating a course."""
title: Optional[str] = Field(None, min_length=1, max_length=255)
description: Optional[str] = None
category_id: Optional[UUID] = None
osha_standard_ref: Optional[str] = Field(None, max_length=50)
estimated_duration_minutes: Optional[int] = Field(None, ge=1)
passing_score_percent: Optional[int] = Field(None, ge=0, le=100)
max_retakes: Optional[int] = Field(None, ge=0)
class CourseListItem(BaseModel):
"""Lightweight course item for list views."""
model_config = ConfigDict(from_attributes=True)
course_id: UUID
title: str
description: Optional[str] = None
category_name: Optional[str] = None
osha_standard_ref: Optional[str] = None
estimated_duration_minutes: Optional[int] = None
status: str
version_number: int
lesson_count: int = 0
quiz_question_count: int = 0
assignment_count: int = 0
created_at: datetime
updated_at: Optional[datetime] = None
class CourseListResponse(BaseModel):
"""Paginated course list response."""
items: List[CourseListItem]
total: int
page: int
page_size: int
total_pages: int
class CourseDetailResponse(BaseModel):
"""Full course detail with lessons and quizzes."""
model_config = ConfigDict(from_attributes=True)
course_id: UUID
company_id: UUID
title: str
description: Optional[str] = None
category_id: Optional[UUID] = None
category_name: Optional[str] = None
osha_standard_ref: Optional[str] = None
estimated_duration_minutes: Optional[int] = None
passing_score_percent: int
max_retakes: int
status: str
version_number: int
locale: str
parent_course_id: Optional[UUID] = None
plan_version_id: Optional[UUID] = None
lessons: List[LessonResponse] = []
quizzes: List["QuizDetailResponse"] = []
created_by: Optional[UUID] = None
created_at: datetime
updated_at: Optional[datetime] = None
# Forward reference for quiz (defined in quiz.py, imported at module level)
from app.schemas.safepath.quiz import QuizDetailResponse # noqa: E402
CourseDetailResponse.model_rebuild()
Quiz Schemas
File: tellus-ehs-hazcom-service/app/schemas/safepath/quiz.py
"""SafePath Quiz Schemas
Pydantic schemas for quiz and question management.
"""
from datetime import datetime
from typing import Optional, List
from uuid import UUID
from pydantic import BaseModel, Field, ConfigDict
# ============================================================================
# Quiz Question Schemas
# ============================================================================
class QuestionOptionCreate(BaseModel):
"""Schema for a single quiz question option."""
id: str = Field(..., description="Option identifier (a, b, c, d...)")
text: str = Field(..., min_length=1)
is_correct: bool = False
class QuestionMatchCreate(BaseModel):
"""Schema for a matching-type question pair."""
id: str
left: str = Field(..., min_length=1)
right: str = Field(..., min_length=1)
is_correct: bool = True
class QuizQuestionCreate(BaseModel):
"""Schema for creating a quiz question."""
question_type: str = Field(..., pattern="^(mcq_single|mcq_multi|true_false|matching)$")
question_text: str = Field(..., min_length=1)
options: List[dict] = Field(..., min_length=2, description="Array of option objects")
explanation: Optional[str] = None
sort_order: int = Field(default=0, ge=0)
locale: str = Field(default="en", pattern="^(en|es)$")
class QuizQuestionUpdate(BaseModel):
"""Schema for updating a quiz question."""
question_type: Optional[str] = Field(None, pattern="^(mcq_single|mcq_multi|true_false|matching)$")
question_text: Optional[str] = Field(None, min_length=1)
options: Optional[List[dict]] = None
explanation: Optional[str] = None
sort_order: Optional[int] = Field(None, ge=0)
class QuizQuestionResponse(BaseModel):
"""Response schema for a quiz question.
Note: is_correct flags are included for course editors
but EXCLUDED in learner-facing endpoints (see delivery service).
"""
model_config = ConfigDict(from_attributes=True)
question_id: UUID
question_type: str
question_text: str
options: List[dict]
explanation: Optional[str] = None
sort_order: int
locale: str
# ============================================================================
# Quiz Schemas
# ============================================================================
class QuizCreate(BaseModel):
"""Schema for creating a quiz."""
title: str = Field(default="Course Quiz", max_length=255)
sort_order: int = Field(default=0, ge=0)
class QuizUpdate(BaseModel):
"""Schema for updating a quiz."""
title: Optional[str] = Field(None, max_length=255)
sort_order: Optional[int] = Field(None, ge=0)
class QuizDetailResponse(BaseModel):
"""Full quiz detail with questions."""
model_config = ConfigDict(from_attributes=True)
quiz_id: UUID
title: str
sort_order: int
questions: List[QuizQuestionResponse] = []
created_at: datetime
updated_at: Optional[datetime] = None
Course Service
File: tellus-ehs-hazcom-service/app/services/safepath/course_service.py
"""SafePath Course Service
Handles course CRUD, lesson management, quiz management,
publishing workflow, and versioning.
"""
import math
from typing import Optional, List, Tuple
from uuid import UUID
from datetime import datetime
from sqlalchemy import func, and_
from sqlalchemy.orm import Session, joinedload
from app.db.models.safepath import (
Course,
CourseCategory,
Lesson,
LessonAsset,
Quiz,
QuizQuestion,
SafePathAuditLog,
)
from app.schemas.safepath.course import (
CourseCreate,
CourseUpdate,
CourseListItem,
LessonCreate,
LessonUpdate,
)
from app.schemas.safepath.quiz import QuizCreate, QuizUpdate, QuizQuestionCreate, QuizQuestionUpdate
class CourseService:
"""Service for managing SafePath training courses."""
def __init__(self, db: Session):
self.db = db
# ================================================================
# Course Categories
# ================================================================
def list_categories(self, company_id: UUID) -> List[CourseCategory]:
"""List all available course categories.
Returns system categories plus company-specific custom categories.
"""
return (
self.db.query(CourseCategory)
.filter(
(CourseCategory.is_system == True) | # noqa: E712
(CourseCategory.company_id == company_id)
)
.order_by(CourseCategory.sort_order)
.all()
)
def create_custom_category(
self, company_id: UUID, name: str, description: Optional[str] = None
) -> CourseCategory:
"""Create a company-specific custom category."""
max_order = (
self.db.query(func.max(CourseCategory.sort_order))
.filter(
(CourseCategory.is_system == True) | # noqa: E712
(CourseCategory.company_id == company_id)
)
.scalar() or 0
)
category = CourseCategory(
name=name,
description=description,
is_system=False,
company_id=company_id,
sort_order=max_order + 1,
)
self.db.add(category)
self.db.flush()
return category
# ================================================================
# Course CRUD
# ================================================================
def create_course(
self, company_id: UUID, user_id: UUID, data: CourseCreate
) -> Course:
"""Create a new course in draft status."""
course = Course(
company_id=company_id,
title=data.title,
description=data.description,
category_id=data.category_id,
osha_standard_ref=data.osha_standard_ref,
estimated_duration_minutes=data.estimated_duration_minutes,
passing_score_percent=data.passing_score_percent,
max_retakes=data.max_retakes,
locale=data.locale,
status="draft",
version_number=1,
created_by=user_id,
)
self.db.add(course)
self.db.flush()
self._log_event(
company_id=company_id,
event_type="course.created",
entity_type="course",
entity_id=course.course_id,
user_id=user_id,
details={"title": data.title},
)
return course
def get_course(self, company_id: UUID, course_id: UUID) -> Optional[Course]:
"""Get a course by ID with lessons and quizzes eagerly loaded."""
return (
self.db.query(Course)
.options(
joinedload(Course.lessons).joinedload(Lesson.assets),
joinedload(Course.quizzes).joinedload(Quiz.questions),
joinedload(Course.category),
)
.filter(
Course.course_id == course_id,
Course.company_id == company_id,
)
.first()
)
def list_courses(
self,
company_id: UUID,
page: int = 1,
page_size: int = 25,
status: Optional[str] = None,
category_id: Optional[UUID] = None,
search: Optional[str] = None,
) -> Tuple[List[CourseListItem], int]:
"""List courses with filtering, search, and pagination.
Returns (items, total_count).
"""
query = self.db.query(Course).filter(Course.company_id == company_id)
if status:
query = query.filter(Course.status == status)
if category_id:
query = query.filter(Course.category_id == category_id)
if search:
search_pattern = f"%{search}%"
query = query.filter(Course.title.ilike(search_pattern))
# Only show latest versions (not parent courses replaced by new versions)
# A course with status='archived' and a child version should be hidden
# unless explicitly filtering archived
if status != "archived":
query = query.filter(Course.status != "archived")
total = query.count()
courses = (
query
.options(joinedload(Course.category))
.order_by(Course.updated_at.desc().nullslast(), Course.created_at.desc())
.offset((page - 1) * page_size)
.limit(page_size)
.all()
)
items = []
for c in courses:
# Count lessons and quiz questions
lesson_count = self.db.query(func.count(Lesson.lesson_id)).filter(
Lesson.course_id == c.course_id
).scalar()
quiz_q_count = (
self.db.query(func.count(QuizQuestion.question_id))
.join(Quiz, Quiz.quiz_id == QuizQuestion.quiz_id)
.filter(Quiz.course_id == c.course_id)
.scalar()
)
from app.db.models.safepath import Assignment
assignment_count = self.db.query(func.count(Assignment.assignment_id)).filter(
Assignment.course_id == c.course_id
).scalar()
items.append(CourseListItem(
course_id=c.course_id,
title=c.title,
description=c.description,
category_name=c.category.name if c.category else None,
osha_standard_ref=c.osha_standard_ref,
estimated_duration_minutes=c.estimated_duration_minutes,
status=c.status,
version_number=c.version_number,
lesson_count=lesson_count,
quiz_question_count=quiz_q_count,
assignment_count=assignment_count,
created_at=c.created_at,
updated_at=c.updated_at,
))
return items, total
def update_course(
self, company_id: UUID, course_id: UUID, user_id: UUID, data: CourseUpdate
) -> Optional[Course]:
"""Update a draft course. Published courses cannot be edited directly
(use create_new_version instead).
"""
course = (
self.db.query(Course)
.filter(Course.course_id == course_id, Course.company_id == company_id)
.first()
)
if not course:
return None
if course.status != "draft":
raise ValueError("Cannot edit a published course. Create a new version instead.")
update_data = data.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(course, field, value)
course.updated_at = datetime.utcnow()
self.db.flush()
self._log_event(
company_id=company_id,
event_type="course.updated",
entity_type="course",
entity_id=course_id,
user_id=user_id,
details={"fields_updated": list(update_data.keys())},
)
return course
def publish_course(
self, company_id: UUID, course_id: UUID, user_id: UUID
) -> Optional[Course]:
"""Publish a draft course, making it available for assignments.
Validates that course has at least one lesson.
"""
course = (
self.db.query(Course)
.filter(Course.course_id == course_id, Course.company_id == company_id)
.first()
)
if not course:
return None
if course.status != "draft":
raise ValueError(f"Course is '{course.status}', only draft courses can be published.")
# Validate: must have at least one lesson
lesson_count = self.db.query(func.count(Lesson.lesson_id)).filter(
Lesson.course_id == course_id
).scalar()
if lesson_count == 0:
raise ValueError("Course must have at least one lesson before publishing.")
course.status = "published"
course.updated_at = datetime.utcnow()
self.db.flush()
self._log_event(
company_id=company_id,
event_type="course.published",
entity_type="course",
entity_id=course_id,
user_id=user_id,
)
return course
def archive_course(
self, company_id: UUID, course_id: UUID, user_id: UUID
) -> Optional[Course]:
"""Archive a course. Existing assignments remain valid."""
course = (
self.db.query(Course)
.filter(Course.course_id == course_id, Course.company_id == company_id)
.first()
)
if not course:
return None
course.status = "archived"
course.updated_at = datetime.utcnow()
self.db.flush()
self._log_event(
company_id=company_id,
event_type="course.archived",
entity_type="course",
entity_id=course_id,
user_id=user_id,
)
return course
def create_new_version(
self, company_id: UUID, course_id: UUID, user_id: UUID
) -> Optional[Course]:
"""Create a new draft version of a published course.
The original course is archived. The new version copies all
lessons and quizzes. Existing assignments stay pinned to the
original version.
"""
original = self.get_course(company_id, course_id)
if not original:
return None
if original.status != "published":
raise ValueError("Can only create new versions of published courses.")
# Create new draft version
new_course = Course(
company_id=company_id,
title=original.title,
description=original.description,
category_id=original.category_id,
osha_standard_ref=original.osha_standard_ref,
estimated_duration_minutes=original.estimated_duration_minutes,
passing_score_percent=original.passing_score_percent,
max_retakes=original.max_retakes,
locale=original.locale,
status="draft",
version_number=original.version_number + 1,
parent_course_id=original.course_id,
plan_version_id=original.plan_version_id,
created_by=user_id,
)
self.db.add(new_course)
self.db.flush()
# Copy lessons
for lesson in original.lessons:
new_lesson = Lesson(
course_id=new_course.course_id,
title=lesson.title,
lesson_type=lesson.lesson_type,
content=lesson.content,
sort_order=lesson.sort_order,
completion_threshold_percent=lesson.completion_threshold_percent,
locale=lesson.locale,
)
self.db.add(new_lesson)
self.db.flush()
# Copy asset references (assets themselves are not duplicated)
for asset in lesson.assets:
new_asset = LessonAsset(
lesson_id=new_lesson.lesson_id,
file_name=asset.file_name,
s3_bucket=asset.s3_bucket,
s3_key=asset.s3_key,
file_type=asset.file_type,
file_size_bytes=asset.file_size_bytes,
)
self.db.add(new_asset)
# Copy quizzes and questions
for quiz in original.quizzes:
new_quiz = Quiz(
course_id=new_course.course_id,
title=quiz.title,
sort_order=quiz.sort_order,
)
self.db.add(new_quiz)
self.db.flush()
for q in quiz.questions:
new_question = QuizQuestion(
quiz_id=new_quiz.quiz_id,
question_type=q.question_type,
question_text=q.question_text,
options=q.options,
explanation=q.explanation,
sort_order=q.sort_order,
locale=q.locale,
)
self.db.add(new_question)
# Archive original
original.status = "archived"
original.updated_at = datetime.utcnow()
self.db.flush()
self._log_event(
company_id=company_id,
event_type="course.versioned",
entity_type="course",
entity_id=new_course.course_id,
user_id=user_id,
details={
"parent_course_id": str(original.course_id),
"version_number": new_course.version_number,
},
)
return new_course
# ================================================================
# Lesson Management
# ================================================================
def add_lesson(
self, company_id: UUID, course_id: UUID, data: LessonCreate
) -> Optional[Lesson]:
"""Add a lesson to a draft course."""
course = (
self.db.query(Course)
.filter(Course.course_id == course_id, Course.company_id == company_id)
.first()
)
if not course:
return None
if course.status != "draft":
raise ValueError("Can only add lessons to draft courses.")
lesson = Lesson(
course_id=course_id,
title=data.title,
lesson_type=data.lesson_type,
content=data.content,
sort_order=data.sort_order,
completion_threshold_percent=data.completion_threshold_percent,
locale=data.locale,
)
self.db.add(lesson)
self.db.flush()
return lesson
def update_lesson(
self, lesson_id: UUID, data: LessonUpdate
) -> Optional[Lesson]:
"""Update a lesson."""
lesson = self.db.query(Lesson).filter(Lesson.lesson_id == lesson_id).first()
if not lesson:
return None
update_data = data.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(lesson, field, value)
lesson.updated_at = datetime.utcnow()
self.db.flush()
return lesson
def delete_lesson(self, lesson_id: UUID) -> bool:
"""Delete a lesson and its assets."""
lesson = self.db.query(Lesson).filter(Lesson.lesson_id == lesson_id).first()
if not lesson:
return False
self.db.delete(lesson)
self.db.flush()
return True
def reorder_lessons(self, course_id: UUID, lesson_ids: List[UUID]) -> bool:
"""Reorder lessons by updating sort_order based on the provided ID sequence."""
for idx, lid in enumerate(lesson_ids):
self.db.query(Lesson).filter(
Lesson.lesson_id == lid, Lesson.course_id == course_id
).update({"sort_order": idx})
self.db.flush()
return True
# ================================================================
# Quiz Management
# ================================================================
def add_quiz(
self, company_id: UUID, course_id: UUID, data: QuizCreate
) -> Optional[Quiz]:
"""Add a quiz to a course."""
course = (
self.db.query(Course)
.filter(Course.course_id == course_id, Course.company_id == company_id)
.first()
)
if not course:
return None
quiz = Quiz(
course_id=course_id,
title=data.title,
sort_order=data.sort_order,
)
self.db.add(quiz)
self.db.flush()
return quiz
def add_question(
self, quiz_id: UUID, data: QuizQuestionCreate
) -> Optional[QuizQuestion]:
"""Add a question to a quiz."""
quiz = self.db.query(Quiz).filter(Quiz.quiz_id == quiz_id).first()
if not quiz:
return None
question = QuizQuestion(
quiz_id=quiz_id,
question_type=data.question_type,
question_text=data.question_text,
options=data.options,
explanation=data.explanation,
sort_order=data.sort_order,
locale=data.locale,
)
self.db.add(question)
self.db.flush()
return question
def update_question(
self, question_id: UUID, data: QuizQuestionUpdate
) -> Optional[QuizQuestion]:
"""Update a quiz question."""
question = self.db.query(QuizQuestion).filter(
QuizQuestion.question_id == question_id
).first()
if not question:
return None
update_data = data.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(question, field, value)
self.db.flush()
return question
def delete_question(self, question_id: UUID) -> bool:
"""Delete a quiz question."""
question = self.db.query(QuizQuestion).filter(
QuizQuestion.question_id == question_id
).first()
if not question:
return False
self.db.delete(question)
self.db.flush()
return True
# ================================================================
# Audit Logging Helper
# ================================================================
def _log_event(
self,
company_id: UUID,
event_type: str,
entity_type: str,
entity_id: UUID,
user_id: Optional[UUID] = None,
details: Optional[dict] = None,
ip_hash: Optional[str] = None,
):
"""Write an entry to the SafePath audit log."""
log = SafePathAuditLog(
company_id=company_id,
event_type=event_type,
entity_type=entity_type,
entity_id=entity_id,
user_id=user_id,
details=details,
ip_hash=ip_hash,
)
self.db.add(log)
API Endpoints
File: tellus-ehs-hazcom-service/app/api/v1/safepath/courses.py
"""SafePath Course API Endpoints
CRUD operations for courses, lessons, and quizzes.
"""
import math
from typing import Optional
from uuid import UUID
from fastapi import APIRouter, Depends, HTTPException, Query, UploadFile, File
from sqlalchemy.orm import Session
from app.db.session import get_db
from app.api.v1.adminhq.auth import get_user_context, UserContext
from app.services.safepath.course_service import CourseService
from app.schemas.safepath.course import (
CourseCreate,
CourseUpdate,
CourseListResponse,
CourseDetailResponse,
CourseCategoryResponse,
LessonCreate,
LessonUpdate,
LessonResponse,
)
from app.schemas.safepath.quiz import (
QuizCreate,
QuizUpdate,
QuizDetailResponse,
QuizQuestionCreate,
QuizQuestionUpdate,
QuizQuestionResponse,
)
router = APIRouter(prefix="/training", tags=["SafePath Training"])
# ============================================================================
# Course Categories
# ============================================================================
@router.get("/categories", response_model=list[CourseCategoryResponse])
def list_categories(
ctx: UserContext = Depends(get_user_context),
db: Session = Depends(get_db),
):
"""List all available course categories (system + custom)."""
service = CourseService(db)
categories = service.list_categories(ctx.company_id)
return [CourseCategoryResponse.model_validate(c) for c in categories]
# ============================================================================
# Course CRUD
# ============================================================================
@router.post("/courses", response_model=CourseDetailResponse, status_code=201)
def create_course(
data: CourseCreate,
ctx: UserContext = Depends(get_user_context),
db: Session = Depends(get_db),
):
"""Create a new training course in draft status."""
service = CourseService(db)
course = service.create_course(ctx.company_id, ctx.user_id, data)
db.commit()
# Re-fetch with relationships loaded
course = service.get_course(ctx.company_id, course.course_id)
return _build_course_detail(course)
@router.get("/courses", response_model=CourseListResponse)
def list_courses(
page: int = Query(1, ge=1),
page_size: int = Query(25, ge=1, le=100),
status: Optional[str] = Query(None, description="draft, published, archived"),
category_id: Optional[UUID] = Query(None),
search: Optional[str] = Query(None),
ctx: UserContext = Depends(get_user_context),
db: Session = Depends(get_db),
):
"""List courses with filtering and pagination."""
service = CourseService(db)
items, total = service.list_courses(
company_id=ctx.company_id,
page=page,
page_size=page_size,
status=status,
category_id=category_id,
search=search,
)
return CourseListResponse(
items=items,
total=total,
page=page,
page_size=page_size,
total_pages=math.ceil(total / page_size) if total > 0 else 0,
)
@router.get("/courses/{course_id}", response_model=CourseDetailResponse)
def get_course(
course_id: UUID,
ctx: UserContext = Depends(get_user_context),
db: Session = Depends(get_db),
):
"""Get full course detail with lessons and quizzes."""
service = CourseService(db)
course = service.get_course(ctx.company_id, course_id)
if not course:
raise HTTPException(status_code=404, detail="Course not found")
return _build_course_detail(course)
@router.put("/courses/{course_id}", response_model=CourseDetailResponse)
def update_course(
course_id: UUID,
data: CourseUpdate,
ctx: UserContext = Depends(get_user_context),
db: Session = Depends(get_db),
):
"""Update a draft course."""
service = CourseService(db)
try:
course = service.update_course(ctx.company_id, course_id, ctx.user_id, data)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
if not course:
raise HTTPException(status_code=404, detail="Course not found")
db.commit()
course = service.get_course(ctx.company_id, course_id)
return _build_course_detail(course)
@router.post("/courses/{course_id}/publish", response_model=CourseDetailResponse)
def publish_course(
course_id: UUID,
ctx: UserContext = Depends(get_user_context),
db: Session = Depends(get_db),
):
"""Publish a draft course."""
service = CourseService(db)
try:
course = service.publish_course(ctx.company_id, course_id, ctx.user_id)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
if not course:
raise HTTPException(status_code=404, detail="Course not found")
db.commit()
course = service.get_course(ctx.company_id, course_id)
return _build_course_detail(course)
@router.post("/courses/{course_id}/archive", response_model=CourseDetailResponse)
def archive_course(
course_id: UUID,
ctx: UserContext = Depends(get_user_context),
db: Session = Depends(get_db),
):
"""Archive a course."""
service = CourseService(db)
course = service.archive_course(ctx.company_id, course_id, ctx.user_id)
if not course:
raise HTTPException(status_code=404, detail="Course not found")
db.commit()
course = service.get_course(ctx.company_id, course_id)
return _build_course_detail(course)
@router.post("/courses/{course_id}/new-version", response_model=CourseDetailResponse)
def create_new_version(
course_id: UUID,
ctx: UserContext = Depends(get_user_context),
db: Session = Depends(get_db),
):
"""Create a new draft version of a published course."""
service = CourseService(db)
try:
new_course = service.create_new_version(ctx.company_id, course_id, ctx.user_id)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
if not new_course:
raise HTTPException(status_code=404, detail="Course not found")
db.commit()
new_course = service.get_course(ctx.company_id, new_course.course_id)
return _build_course_detail(new_course)
# ============================================================================
# Lesson Endpoints
# ============================================================================
@router.post("/courses/{course_id}/lessons", response_model=LessonResponse, status_code=201)
def add_lesson(
course_id: UUID,
data: LessonCreate,
ctx: UserContext = Depends(get_user_context),
db: Session = Depends(get_db),
):
"""Add a lesson to a draft course."""
service = CourseService(db)
try:
lesson = service.add_lesson(ctx.company_id, course_id, data)
except ValueError as e:
raise HTTPException(status_code=400, detail=str(e))
if not lesson:
raise HTTPException(status_code=404, detail="Course not found")
db.commit()
return LessonResponse.model_validate(lesson)
@router.put("/lessons/{lesson_id}", response_model=LessonResponse)
def update_lesson(
lesson_id: UUID,
data: LessonUpdate,
ctx: UserContext = Depends(get_user_context),
db: Session = Depends(get_db),
):
"""Update a lesson."""
service = CourseService(db)
lesson = service.update_lesson(lesson_id, data)
if not lesson:
raise HTTPException(status_code=404, detail="Lesson not found")
db.commit()
return LessonResponse.model_validate(lesson)
@router.delete("/lessons/{lesson_id}", status_code=204)
def delete_lesson(
lesson_id: UUID,
ctx: UserContext = Depends(get_user_context),
db: Session = Depends(get_db),
):
"""Delete a lesson."""
service = CourseService(db)
if not service.delete_lesson(lesson_id):
raise HTTPException(status_code=404, detail="Lesson not found")
db.commit()
# ============================================================================
# Quiz Endpoints
# ============================================================================
@router.post("/courses/{course_id}/quizzes", response_model=QuizDetailResponse, status_code=201)
def add_quiz(
course_id: UUID,
data: QuizCreate,
ctx: UserContext = Depends(get_user_context),
db: Session = Depends(get_db),
):
"""Add a quiz to a course."""
service = CourseService(db)
quiz = service.add_quiz(ctx.company_id, course_id, data)
if not quiz:
raise HTTPException(status_code=404, detail="Course not found")
db.commit()
return QuizDetailResponse.model_validate(quiz)
@router.post("/quizzes/{quiz_id}/questions", response_model=QuizQuestionResponse, status_code=201)
def add_question(
quiz_id: UUID,
data: QuizQuestionCreate,
ctx: UserContext = Depends(get_user_context),
db: Session = Depends(get_db),
):
"""Add a question to a quiz."""
service = CourseService(db)
question = service.add_question(quiz_id, data)
if not question:
raise HTTPException(status_code=404, detail="Quiz not found")
db.commit()
return QuizQuestionResponse.model_validate(question)
@router.put("/questions/{question_id}", response_model=QuizQuestionResponse)
def update_question(
question_id: UUID,
data: QuizQuestionUpdate,
ctx: UserContext = Depends(get_user_context),
db: Session = Depends(get_db),
):
"""Update a quiz question."""
service = CourseService(db)
question = service.update_question(question_id, data)
if not question:
raise HTTPException(status_code=404, detail="Question not found")
db.commit()
return QuizQuestionResponse.model_validate(question)
@router.delete("/questions/{question_id}", status_code=204)
def delete_question(
question_id: UUID,
ctx: UserContext = Depends(get_user_context),
db: Session = Depends(get_db),
):
"""Delete a quiz question."""
service = CourseService(db)
if not service.delete_question(question_id):
raise HTTPException(status_code=404, detail="Question not found")
db.commit()
# ============================================================================
# Helper
# ============================================================================
def _build_course_detail(course) -> CourseDetailResponse:
"""Build CourseDetailResponse from a Course model with loaded relationships."""
from app.schemas.safepath.quiz import QuizDetailResponse, QuizQuestionResponse
quizzes = []
for q in course.quizzes:
questions = [
QuizQuestionResponse.model_validate(qq) for qq in q.questions
]
quizzes.append(QuizDetailResponse(
quiz_id=q.quiz_id,
title=q.title,
sort_order=q.sort_order,
questions=questions,
created_at=q.created_at,
updated_at=q.updated_at,
))
lessons = [LessonResponse.model_validate(l) for l in course.lessons]
return CourseDetailResponse(
course_id=course.course_id,
company_id=course.company_id,
title=course.title,
description=course.description,
category_id=course.category_id,
category_name=course.category.name if course.category else None,
osha_standard_ref=course.osha_standard_ref,
estimated_duration_minutes=course.estimated_duration_minutes,
passing_score_percent=course.passing_score_percent,
max_retakes=course.max_retakes,
status=course.status,
version_number=course.version_number,
locale=course.locale,
parent_course_id=course.parent_course_id,
plan_version_id=course.plan_version_id,
lessons=lessons,
quizzes=quizzes,
created_by=course.created_by,
created_at=course.created_at,
updated_at=course.updated_at,
)
Register Router
File: tellus-ehs-hazcom-service/app/api/v1/safepath/__init__.py
from fastapi import APIRouter
from app.api.v1.safepath.courses import router as courses_router
router = APIRouter()
router.include_router(courses_router)
Then in the main app or router registry, add:
from app.api.v1.safepath import router as safepath_router
app.include_router(safepath_router, prefix="/api/v1")
Frontend TypeScript Types
File: tellus-ehs-hazcom-ui/src/types/safepath.ts
// SafePath Training TypeScript interfaces
export interface CourseCategory {
category_id: string;
name: string;
description: string | null;
is_system: boolean;
sort_order: number;
}
export interface LessonAsset {
asset_id: string;
file_name: string;
file_type: string;
file_size_bytes: number | null;
download_url: string | null;
}
export interface Lesson {
lesson_id: string;
title: string;
lesson_type: 'video' | 'pdf' | 'slides' | 'text' | 'external_link';
content: Record<string, any> | null;
sort_order: number;
completion_threshold_percent: number;
locale: string;
assets: LessonAsset[];
created_at: string;
}
export interface QuizQuestion {
question_id: string;
question_type: 'mcq_single' | 'mcq_multi' | 'true_false' | 'matching';
question_text: string;
options: Array<{
id: string;
text: string;
is_correct?: boolean; // Only present for course editors, not learners
}>;
explanation: string | null;
sort_order: number;
locale: string;
}
export interface Quiz {
quiz_id: string;
title: string;
sort_order: number;
questions: QuizQuestion[];
created_at: string;
updated_at: string | null;
}
export interface CourseListItem {
course_id: string;
title: string;
description: string | null;
category_name: string | null;
osha_standard_ref: string | null;
estimated_duration_minutes: number | null;
status: 'draft' | 'published' | 'archived';
version_number: number;
lesson_count: number;
quiz_question_count: number;
assignment_count: number;
created_at: string;
updated_at: string | null;
}
export interface CourseDetail extends CourseListItem {
company_id: string;
category_id: string | null;
passing_score_percent: number;
max_retakes: number;
locale: string;
parent_course_id: string | null;
plan_version_id: string | null;
lessons: Lesson[];
quizzes: Quiz[];
created_by: string | null;
}
export interface CourseListResponse {
items: CourseListItem[];
total: number;
page: number;
page_size: number;
total_pages: number;
}
// --- Create/Update payloads ---
export interface CourseCreatePayload {
title: string;
description?: string;
category_id?: string;
osha_standard_ref?: string;
estimated_duration_minutes?: number;
passing_score_percent?: number;
max_retakes?: number;
locale?: string;
}
export interface LessonCreatePayload {
title: string;
lesson_type: 'video' | 'pdf' | 'slides' | 'text' | 'external_link';
content?: Record<string, any>;
sort_order?: number;
completion_threshold_percent?: number;
locale?: string;
}
export interface QuizQuestionCreatePayload {
question_type: 'mcq_single' | 'mcq_multi' | 'true_false' | 'matching';
question_text: string;
options: Array<{ id: string; text: string; is_correct: boolean }>;
explanation?: string;
sort_order?: number;
locale?: string;
}
Frontend API Service
File: tellus-ehs-hazcom-ui/src/services/safepath-api.ts
import api from './api';
import type {
CourseCategory,
CourseListResponse,
CourseDetail,
CourseCreatePayload,
LessonCreatePayload,
Lesson,
Quiz,
QuizQuestion,
QuizQuestionCreatePayload,
} from '../types/safepath';
const BASE = '/training';
// Categories
export const getCategories = () =>
api.get<CourseCategory[]>(`${BASE}/categories`).then(r => r.data);
// Courses
export const getCourses = (params: {
page?: number;
page_size?: number;
status?: string;
category_id?: string;
search?: string;
}) =>
api.get<CourseListResponse>(`${BASE}/courses`, { params }).then(r => r.data);
export const getCourse = (courseId: string) =>
api.get<CourseDetail>(`${BASE}/courses/${courseId}`).then(r => r.data);
export const createCourse = (data: CourseCreatePayload) =>
api.post<CourseDetail>(`${BASE}/courses`, data).then(r => r.data);
export const updateCourse = (courseId: string, data: Partial<CourseCreatePayload>) =>
api.put<CourseDetail>(`${BASE}/courses/${courseId}`, data).then(r => r.data);
export const publishCourse = (courseId: string) =>
api.post<CourseDetail>(`${BASE}/courses/${courseId}/publish`).then(r => r.data);
export const archiveCourse = (courseId: string) =>
api.post<CourseDetail>(`${BASE}/courses/${courseId}/archive`).then(r => r.data);
export const createNewVersion = (courseId: string) =>
api.post<CourseDetail>(`${BASE}/courses/${courseId}/new-version`).then(r => r.data);
// Lessons
export const addLesson = (courseId: string, data: LessonCreatePayload) =>
api.post<Lesson>(`${BASE}/courses/${courseId}/lessons`, data).then(r => r.data);
export const updateLesson = (lessonId: string, data: Partial<LessonCreatePayload>) =>
api.put<Lesson>(`${BASE}/lessons/${lessonId}`, data).then(r => r.data);
export const deleteLesson = (lessonId: string) =>
api.delete(`${BASE}/lessons/${lessonId}`);
// Quizzes
export const addQuiz = (courseId: string, data?: { title?: string }) =>
api.post<Quiz>(`${BASE}/courses/${courseId}/quizzes`, data || {}).then(r => r.data);
export const addQuestion = (quizId: string, data: QuizQuestionCreatePayload) =>
api.post<QuizQuestion>(`${BASE}/quizzes/${quizId}/questions`, data).then(r => r.data);
export const updateQuestion = (questionId: string, data: Partial<QuizQuestionCreatePayload>) =>
api.put<QuizQuestion>(`${BASE}/questions/${questionId}`, data).then(r => r.data);
export const deleteQuestion = (questionId: string) =>
api.delete(`${BASE}/questions/${questionId}`);
Verification Checklist
After implementing this phase:
-
Backend:
GET /api/v1/training/categoriesreturns 16 system categoriesPOST /api/v1/training/coursescreates a draft courseGET /api/v1/training/courseslists courses with paginationGET /api/v1/training/courses/:idreturns full detail with lessons and quizzesPUT /api/v1/training/courses/:idupdates a draft course (fails for published)POST /api/v1/training/courses/:id/publishpublishes (fails if no lessons)POST /api/v1/training/courses/:id/archivearchives a coursePOST /api/v1/training/courses/:id/new-versioncreates a new draft version- Lesson CRUD works on draft courses only
- Quiz and question CRUD works
-
Frontend:
- Course list page shows all courses with status badges
- Create course wizard allows step-by-step creation
- Course detail page shows lessons and quizzes
- Publish/archive actions work from detail page