321 lines
11 KiB
Python
321 lines
11 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
||
|
|
from sqlalchemy import or_, select
|
||
|
|
from sqlalchemy.orm import Session, selectinload
|
||
|
|
|
||
|
|
from app.auth import get_active_user, get_consented_user, get_optional_user
|
||
|
|
from app.database import get_db
|
||
|
|
from app.models import Question, Quiz, QuizAttempt, QuizVote, User
|
||
|
|
from app.schemas import (
|
||
|
|
QuestionAuthorOut,
|
||
|
|
QuestionPlayOut,
|
||
|
|
QuestionResultOut,
|
||
|
|
QuizCreateIn,
|
||
|
|
QuizDetailOut,
|
||
|
|
QuizSubmitIn,
|
||
|
|
QuizSubmitOut,
|
||
|
|
QuizSummaryOut,
|
||
|
|
QuizUpdateIn,
|
||
|
|
VoteIn,
|
||
|
|
VoteOut,
|
||
|
|
)
|
||
|
|
from app.services.quiz_service import (
|
||
|
|
compute_quiz_mode,
|
||
|
|
enrich_quizzes,
|
||
|
|
generate_share_slug,
|
||
|
|
grade_question,
|
||
|
|
serialize_play_options,
|
||
|
|
)
|
||
|
|
|
||
|
|
router = APIRouter(prefix="/api/v1/quizzes", tags=["Quizzes"])
|
||
|
|
|
||
|
|
|
||
|
|
def _load_quiz_by_slug_or_id(db: Session, slug_or_id: str) -> Quiz | None:
|
||
|
|
stmt = select(Quiz).options(selectinload(Quiz.author), selectinload(Quiz.questions))
|
||
|
|
if slug_or_id.isdigit():
|
||
|
|
quiz = db.scalar(stmt.where(Quiz.id == int(slug_or_id)))
|
||
|
|
if quiz is not None:
|
||
|
|
return quiz
|
||
|
|
return db.scalar(stmt.where(Quiz.share_slug == slug_or_id))
|
||
|
|
|
||
|
|
|
||
|
|
@router.get("", response_model=list[QuizSummaryOut])
|
||
|
|
def list_quizzes(
|
||
|
|
q: str | None = Query(default=None, max_length=120),
|
||
|
|
mode: str | None = Query(default=None),
|
||
|
|
mine_only: bool = Query(default=False),
|
||
|
|
db: Session = Depends(get_db),
|
||
|
|
current_user: User | None = Depends(get_optional_user),
|
||
|
|
) -> list[QuizSummaryOut]:
|
||
|
|
stmt = (
|
||
|
|
select(Quiz)
|
||
|
|
.options(selectinload(Quiz.author), selectinload(Quiz.questions))
|
||
|
|
.where(Quiz.is_removed.is_(False))
|
||
|
|
)
|
||
|
|
if mine_only:
|
||
|
|
if current_user is None:
|
||
|
|
raise HTTPException(
|
||
|
|
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
|
|
detail="Sign in to view your own quizzes.",
|
||
|
|
)
|
||
|
|
stmt = stmt.where(Quiz.author_id == current_user.id)
|
||
|
|
elif current_user is not None:
|
||
|
|
stmt = stmt.where(or_(Quiz.is_public.is_(True), Quiz.author_id == current_user.id))
|
||
|
|
else:
|
||
|
|
stmt = stmt.where(Quiz.is_public.is_(True))
|
||
|
|
|
||
|
|
if q and q.strip():
|
||
|
|
needle = f"%{q.strip()}%"
|
||
|
|
stmt = stmt.where(or_(Quiz.title.ilike(needle), Quiz.description.ilike(needle)))
|
||
|
|
|
||
|
|
stmt = stmt.order_by(Quiz.created_at.desc(), Quiz.id.desc())
|
||
|
|
quizzes = list(db.scalars(stmt).all())
|
||
|
|
|
||
|
|
if mode and mode not in ("all", ""):
|
||
|
|
quizzes = [quiz for quiz in quizzes if compute_quiz_mode(quiz.questions) == mode]
|
||
|
|
|
||
|
|
return enrich_quizzes(db, quizzes, current_user)
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("", response_model=QuizDetailOut, status_code=status.HTTP_201_CREATED)
|
||
|
|
def create_quiz(
|
||
|
|
payload: QuizCreateIn,
|
||
|
|
db: Session = Depends(get_db),
|
||
|
|
user: User = Depends(get_consented_user),
|
||
|
|
) -> QuizDetailOut:
|
||
|
|
if len(payload.questions) > 40:
|
||
|
|
raise HTTPException(
|
||
|
|
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||
|
|
detail="Public free tier quizzes may contain at most 40 questions.",
|
||
|
|
)
|
||
|
|
slug = generate_share_slug(db, payload.title)
|
||
|
|
quiz = Quiz(
|
||
|
|
share_slug=slug,
|
||
|
|
author_id=user.id,
|
||
|
|
title=payload.title,
|
||
|
|
description=payload.description.strip(),
|
||
|
|
is_public=payload.is_public,
|
||
|
|
)
|
||
|
|
for idx, q_in in enumerate(payload.questions):
|
||
|
|
quiz.questions.append(
|
||
|
|
Question(
|
||
|
|
position=idx,
|
||
|
|
question_type=q_in.question_type,
|
||
|
|
prompt=q_in.prompt,
|
||
|
|
options=[opt.model_dump() for opt in q_in.options],
|
||
|
|
correct_answer=q_in.correct_answer,
|
||
|
|
accepted_answers=q_in.accepted_answers,
|
||
|
|
)
|
||
|
|
)
|
||
|
|
db.add(quiz)
|
||
|
|
db.commit()
|
||
|
|
loaded = _load_quiz_by_slug_or_id(db, str(quiz.id))
|
||
|
|
assert loaded is not None
|
||
|
|
return _build_quiz_detail(db, loaded, user)
|
||
|
|
|
||
|
|
|
||
|
|
def _build_quiz_detail(db: Session, quiz: Quiz, current_user: User | None) -> QuizDetailOut:
|
||
|
|
summary = enrich_quizzes(db, [quiz], current_user)[0]
|
||
|
|
play_questions = [
|
||
|
|
QuestionPlayOut(
|
||
|
|
id=q.id,
|
||
|
|
position=q.position,
|
||
|
|
question_type=q.question_type, # type: ignore[arg-type]
|
||
|
|
prompt=q.prompt,
|
||
|
|
options=serialize_play_options(q.options),
|
||
|
|
)
|
||
|
|
for q in quiz.questions
|
||
|
|
]
|
||
|
|
editable: list[QuestionAuthorOut] | None = None
|
||
|
|
if current_user and (current_user.id == quiz.author_id or current_user.is_admin):
|
||
|
|
editable = [
|
||
|
|
QuestionAuthorOut(
|
||
|
|
id=q.id,
|
||
|
|
position=q.position,
|
||
|
|
question_type=q.question_type, # type: ignore[arg-type]
|
||
|
|
prompt=q.prompt,
|
||
|
|
options=serialize_play_options(q.options),
|
||
|
|
correct_answer=q.correct_answer,
|
||
|
|
accepted_answers=list(q.accepted_answers or []),
|
||
|
|
)
|
||
|
|
for q in quiz.questions
|
||
|
|
]
|
||
|
|
return QuizDetailOut(
|
||
|
|
**summary.model_dump(),
|
||
|
|
questions=play_questions,
|
||
|
|
editable_questions=editable,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
@router.get("/{slug_or_id}", response_model=QuizDetailOut)
|
||
|
|
def get_quiz(
|
||
|
|
slug_or_id: str,
|
||
|
|
db: Session = Depends(get_db),
|
||
|
|
current_user: User | None = Depends(get_optional_user),
|
||
|
|
) -> QuizDetailOut:
|
||
|
|
quiz = _load_quiz_by_slug_or_id(db, slug_or_id)
|
||
|
|
if quiz is None or quiz.is_removed:
|
||
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Quiz not found.")
|
||
|
|
if not quiz.is_public:
|
||
|
|
if current_user is None or (
|
||
|
|
current_user.id != quiz.author_id and not current_user.is_admin
|
||
|
|
):
|
||
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Quiz not found.")
|
||
|
|
return _build_quiz_detail(db, quiz, current_user)
|
||
|
|
|
||
|
|
|
||
|
|
@router.put("/{quiz_id}", response_model=QuizDetailOut)
|
||
|
|
def update_quiz(
|
||
|
|
quiz_id: int,
|
||
|
|
payload: QuizUpdateIn,
|
||
|
|
db: Session = Depends(get_db),
|
||
|
|
user: User = Depends(get_consented_user),
|
||
|
|
) -> QuizDetailOut:
|
||
|
|
quiz = _load_quiz_by_slug_or_id(db, str(quiz_id))
|
||
|
|
if quiz is None or quiz.is_removed:
|
||
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Quiz not found.")
|
||
|
|
if quiz.author_id != user.id and not user.is_admin:
|
||
|
|
raise HTTPException(
|
||
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
||
|
|
detail="Only the quiz author or a moderator can edit this quiz.",
|
||
|
|
)
|
||
|
|
|
||
|
|
quiz.title = payload.title
|
||
|
|
quiz.description = payload.description.strip()
|
||
|
|
quiz.is_public = payload.is_public
|
||
|
|
quiz.questions.clear()
|
||
|
|
db.flush()
|
||
|
|
|
||
|
|
for idx, q_in in enumerate(payload.questions):
|
||
|
|
quiz.questions.append(
|
||
|
|
Question(
|
||
|
|
position=idx,
|
||
|
|
question_type=q_in.question_type,
|
||
|
|
prompt=q_in.prompt,
|
||
|
|
options=[opt.model_dump() for opt in q_in.options],
|
||
|
|
correct_answer=q_in.correct_answer,
|
||
|
|
accepted_answers=q_in.accepted_answers,
|
||
|
|
)
|
||
|
|
)
|
||
|
|
db.commit()
|
||
|
|
loaded = _load_quiz_by_slug_or_id(db, str(quiz.id))
|
||
|
|
assert loaded is not None
|
||
|
|
return _build_quiz_detail(db, loaded, user)
|
||
|
|
|
||
|
|
|
||
|
|
@router.delete("/{quiz_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||
|
|
def delete_quiz(
|
||
|
|
quiz_id: int,
|
||
|
|
db: Session = Depends(get_db),
|
||
|
|
user: User = Depends(get_active_user),
|
||
|
|
) -> None:
|
||
|
|
quiz = db.get(Quiz, quiz_id)
|
||
|
|
if quiz is None or quiz.is_removed:
|
||
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Quiz not found.")
|
||
|
|
if quiz.author_id != user.id and not user.is_admin:
|
||
|
|
raise HTTPException(
|
||
|
|
status_code=status.HTTP_403_FORBIDDEN,
|
||
|
|
detail="Only the quiz author or a moderator can delete this quiz.",
|
||
|
|
)
|
||
|
|
quiz.is_removed = True
|
||
|
|
db.commit()
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/{quiz_id}/submit", response_model=QuizSubmitOut)
|
||
|
|
def submit_quiz(
|
||
|
|
quiz_id: int,
|
||
|
|
payload: QuizSubmitIn,
|
||
|
|
db: Session = Depends(get_db),
|
||
|
|
user: User = Depends(get_active_user),
|
||
|
|
) -> QuizSubmitOut:
|
||
|
|
quiz = _load_quiz_by_slug_or_id(db, str(quiz_id))
|
||
|
|
if quiz is None or quiz.is_removed:
|
||
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Quiz not found.")
|
||
|
|
if not quiz.is_public and quiz.author_id != user.id and not user.is_admin:
|
||
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Quiz not found.")
|
||
|
|
|
||
|
|
total = len(quiz.questions)
|
||
|
|
correct_count = 0
|
||
|
|
results: list[QuestionResultOut] = []
|
||
|
|
|
||
|
|
for q in quiz.questions:
|
||
|
|
submitted = payload.answers.get(str(q.id), "")
|
||
|
|
is_correct, correct_display = grade_question(q, submitted)
|
||
|
|
if is_correct:
|
||
|
|
correct_count += 1
|
||
|
|
results.append(
|
||
|
|
QuestionResultOut(
|
||
|
|
question_id=q.id,
|
||
|
|
prompt=q.prompt,
|
||
|
|
question_type=q.question_type, # type: ignore[arg-type]
|
||
|
|
submitted_answer=submitted,
|
||
|
|
is_correct=is_correct,
|
||
|
|
correct_display=correct_display,
|
||
|
|
)
|
||
|
|
)
|
||
|
|
|
||
|
|
score_percent = round((correct_count / total) * 100.0, 1) if total > 0 else 100.0
|
||
|
|
attempt = QuizAttempt(
|
||
|
|
quiz_id=quiz.id,
|
||
|
|
user_id=user.id,
|
||
|
|
score_count=correct_count,
|
||
|
|
total_questions=total,
|
||
|
|
score_percent=score_percent,
|
||
|
|
answers_payload=payload.answers,
|
||
|
|
)
|
||
|
|
db.add(attempt)
|
||
|
|
db.commit()
|
||
|
|
db.refresh(attempt)
|
||
|
|
|
||
|
|
summary = enrich_quizzes(db, [quiz], user)[0]
|
||
|
|
assert summary.my_latest_score is not None
|
||
|
|
assert summary.my_best_score is not None
|
||
|
|
assert summary.global_best_score is not None
|
||
|
|
|
||
|
|
return QuizSubmitOut(
|
||
|
|
attempt_id=attempt.id,
|
||
|
|
score_count=correct_count,
|
||
|
|
total_questions=total,
|
||
|
|
score_percent=score_percent,
|
||
|
|
results=results,
|
||
|
|
my_latest_score=summary.my_latest_score,
|
||
|
|
my_best_score=summary.my_best_score,
|
||
|
|
global_best_score=summary.global_best_score,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
@router.post("/{quiz_id}/vote", response_model=VoteOut)
|
||
|
|
def vote_quiz(
|
||
|
|
quiz_id: int,
|
||
|
|
payload: VoteIn,
|
||
|
|
db: Session = Depends(get_db),
|
||
|
|
user: User = Depends(get_consented_user),
|
||
|
|
) -> VoteOut:
|
||
|
|
quiz = db.get(Quiz, quiz_id)
|
||
|
|
if quiz is None or quiz.is_removed or not quiz.is_public:
|
||
|
|
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Public quiz not found.")
|
||
|
|
|
||
|
|
existing = db.scalar(
|
||
|
|
select(QuizVote).where(QuizVote.quiz_id == quiz_id, QuizVote.user_id == user.id)
|
||
|
|
)
|
||
|
|
if payload.value == 0:
|
||
|
|
if existing is not None:
|
||
|
|
db.delete(existing)
|
||
|
|
else:
|
||
|
|
if existing is None:
|
||
|
|
db.add(QuizVote(quiz_id=quiz_id, user_id=user.id, value=payload.value))
|
||
|
|
else:
|
||
|
|
existing.value = payload.value
|
||
|
|
db.commit()
|
||
|
|
|
||
|
|
votes = db.scalars(select(QuizVote).where(QuizVote.quiz_id == quiz_id)).all()
|
||
|
|
upvotes = sum(1 for v in votes if v.value > 0)
|
||
|
|
downvotes = sum(1 for v in votes if v.value < 0)
|
||
|
|
return VoteOut(
|
||
|
|
quiz_id=quiz_id,
|
||
|
|
upvotes=upvotes,
|
||
|
|
downvotes=downvotes,
|
||
|
|
user_vote=payload.value,
|
||
|
|
)
|