2cbbfe00c3
- Fix tag filtering (sa_text import shadowing caused UnboundLocalError) - Add TagBrowser component with per-section search - Multi-category selection (OR within categories, AND with tags) - AI image validation: has_figure field in extraction prompt - Skip known branding images by MD5 hash + dimension filters - Fix quiz timer auto-submit (wrong useEffect dependency) - Fix QuizResponse schema: section_id nullable - Fix Question.quiz_id → source_quiz_id attribute name - Fix SQL injection in quizzes.py vector search - Add PDF processing progress steps via Redis - Add delete user from admin panel - Admin page: no spinner flash on data refresh - Upload progress: axios 1.x e.progress, remove manual Content-Type - Duplicate model error: 409 with clear message - Backend startup: retry DDL migration on lock timeout - Replace all silent except:pass with warning logs - Comprehensive multi-page documentation (docs/) Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
560 lines
21 KiB
Python
560 lines
21 KiB
Python
import random
|
|
from datetime import datetime
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy import cast, String, or_, and_, func
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.database import get_db
|
|
from app.models.quiz import Quiz
|
|
from app.models.question import Question as QuestionModel
|
|
from app.models.section import Section
|
|
from app.models.attempt import QuizAttempt
|
|
from app.models.user import User
|
|
from app.schemas.quiz import QuizCreate, QuizUpdate, QuizResponse, QuizDetail, QuizLearningDetail, QuizReview
|
|
from app.services import quiz_service
|
|
from app.utils.auth import get_current_user, require_moderator
|
|
from app.utils.quiz_questions import get_quiz_questions, question_in_quiz, remove_question_from_quiz
|
|
|
|
router = APIRouter()
|
|
|
|
|
|
@router.post("/")
|
|
def create_quiz(
|
|
quiz_data: QuizCreate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_moderator),
|
|
):
|
|
"""Start async quiz extraction. Returns {job_id} immediately; poll /quizzes/job/{job_id} for progress."""
|
|
import uuid
|
|
section = db.query(Section).filter(Section.id == quiz_data.section_id).first()
|
|
if not section:
|
|
raise HTTPException(status_code=404, detail="Section not found")
|
|
if not current_user.is_admin and section.document.user_id != current_user.id:
|
|
raise HTTPException(status_code=403, detail="Not your document")
|
|
if quiz_data.mode not in ("timed", "learning"):
|
|
raise HTTPException(status_code=400, detail="Mode must be 'timed' or 'learning'")
|
|
|
|
job_id = str(uuid.uuid4())
|
|
|
|
try:
|
|
from app.tasks.quiz_tasks import extract_quiz
|
|
import redis as redis_lib
|
|
from app.config import settings
|
|
r = redis_lib.from_url(settings.REDIS_URL, decode_responses=True)
|
|
r.set(f"extraction:status:{job_id}", "pending", ex=3600)
|
|
# Store job under user so any session can query it
|
|
title = quiz_data.title
|
|
r.lpush(f"extraction:user_jobs:{current_user.id}", job_id)
|
|
r.expire(f"extraction:user_jobs:{current_user.id}", 86400)
|
|
r.set(f"extraction:job_title:{job_id}", title, ex=3600)
|
|
|
|
extract_quiz.delay(
|
|
job_id=job_id,
|
|
user_id=current_user.id,
|
|
section_id=quiz_data.section_id,
|
|
title=quiz_data.title,
|
|
mode=quiz_data.mode,
|
|
time_limit_minutes=quiz_data.time_limit_minutes,
|
|
model_id=quiz_data.model_id,
|
|
question_category_id=quiz_data.question_category_id,
|
|
extraction_mode=quiz_data.extraction_mode,
|
|
)
|
|
except Exception:
|
|
# Celery/Redis unavailable — fall back to synchronous extraction
|
|
try:
|
|
quiz = quiz_service.create_quiz_from_section(
|
|
db=db, user_id=current_user.id,
|
|
section_id=quiz_data.section_id, title=quiz_data.title,
|
|
mode=quiz_data.mode, time_limit_minutes=quiz_data.time_limit_minutes,
|
|
model_id=quiz_data.model_id, question_category_id=quiz_data.question_category_id,
|
|
)
|
|
return {"job_id": job_id, "status": "completed", "quiz_id": quiz.id}
|
|
except (ValueError, RuntimeError) as e:
|
|
raise HTTPException(status_code=400, detail=str(e))
|
|
|
|
return {"job_id": job_id, "status": "pending"}
|
|
|
|
|
|
@router.get("/jobs")
|
|
def list_user_jobs(current_user: User = Depends(get_current_user)):
|
|
"""Return all recent extraction jobs for the logged-in user (any browser/session)."""
|
|
import json as _json
|
|
import redis as redis_lib
|
|
from app.config import settings
|
|
|
|
r = redis_lib.from_url(settings.REDIS_URL, decode_responses=True)
|
|
job_ids = r.lrange(f"extraction:user_jobs:{current_user.id}", 0, 19)
|
|
jobs = []
|
|
for jid in job_ids:
|
|
status = r.get(f"extraction:status:{jid}")
|
|
if not status:
|
|
continue
|
|
steps = r.lrange(f"extraction:steps:{jid}", 0, -1)
|
|
last_step = ""
|
|
if steps:
|
|
last_step = _json.loads(steps[-1]).get("message", "")
|
|
quiz_id = r.get(f"extraction:quiz_id:{jid}")
|
|
title = r.get(f"extraction:job_title:{jid}") or "Quiz"
|
|
jobs.append({
|
|
"job_id": jid,
|
|
"title": title,
|
|
"status": status,
|
|
"steps_count": len(steps),
|
|
"last_step": last_step[:80],
|
|
"quiz_id": int(quiz_id) if quiz_id else None,
|
|
})
|
|
return jobs
|
|
|
|
|
|
@router.get("/job/{job_id}")
|
|
def get_extraction_job(job_id: str, current_user: User = Depends(require_moderator)):
|
|
"""Poll extraction job progress. Returns steps list, status, and quiz_id when done."""
|
|
import json as _json
|
|
import redis as redis_lib
|
|
from app.config import settings
|
|
|
|
r = redis_lib.from_url(settings.REDIS_URL, decode_responses=True)
|
|
status = r.get(f"extraction:status:{job_id}") or "unknown"
|
|
raw_steps = r.lrange(f"extraction:steps:{job_id}", 0, -1)
|
|
steps = [_json.loads(s) for s in raw_steps]
|
|
result = {"job_id": job_id, "status": status, "steps": steps}
|
|
if status == "completed":
|
|
result["quiz_id"] = int(r.get(f"extraction:quiz_id:{job_id}") or 0)
|
|
if status == "failed":
|
|
result["error"] = r.get(f"extraction:error:{job_id}") or "Unknown error"
|
|
return result
|
|
|
|
|
|
@router.post("/job/{job_id}/cancel")
|
|
def cancel_extraction_job(job_id: str, current_user: User = Depends(get_current_user)):
|
|
"""Cancel a running extraction job."""
|
|
import redis as redis_lib
|
|
from app.config import settings
|
|
r = redis_lib.from_url(settings.REDIS_URL, decode_responses=True)
|
|
# Verify this job belongs to the user
|
|
user_jobs = r.lrange(f"extraction:user_jobs:{current_user.id}", 0, 19)
|
|
if job_id not in user_jobs:
|
|
raise HTTPException(status_code=403, detail="Job not found")
|
|
status = r.get(f"extraction:status:{job_id}")
|
|
if status != "running":
|
|
raise HTTPException(status_code=400, detail=f"Job is not running (status: {status})")
|
|
r.set(f"extraction:status:{job_id}", "cancelled")
|
|
return {"status": "cancelled"}
|
|
|
|
|
|
@router.get("/search")
|
|
def search_quizzes(
|
|
q: str = Query(..., min_length=2, max_length=200),
|
|
mode: str = Query("all"), # "title" | "questions" | "all"
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Hybrid semantic + keyword search across quiz titles and questions."""
|
|
from sqlalchemy import text as sa_text
|
|
from app.services import embedding_service
|
|
|
|
phrase = q.strip()
|
|
if not phrase:
|
|
return []
|
|
|
|
results = {} # quiz_id -> result dict
|
|
|
|
def _ensure_quiz(quiz_id: int, match_type: str):
|
|
if quiz_id not in results:
|
|
quiz = db.query(Quiz).filter(Quiz.id == quiz_id).first()
|
|
if not quiz:
|
|
return False
|
|
results[quiz_id] = {
|
|
"quiz_id": quiz.id,
|
|
"quiz_title": quiz.title,
|
|
"questions_count": quiz.questions_count,
|
|
"mode": quiz.mode,
|
|
"time_limit_minutes": quiz.time_limit_minutes,
|
|
"match_type": match_type,
|
|
"matching_questions": [],
|
|
}
|
|
elif results[quiz_id]["match_type"] != match_type and match_type != "title":
|
|
results[quiz_id]["match_type"] = "both"
|
|
return True
|
|
|
|
# ── Title search ─────────────────────────────────────────────
|
|
if mode in ("title", "all"):
|
|
for quiz in db.query(Quiz).filter(Quiz.title.ilike(f"%{phrase}%")).limit(30).all():
|
|
_ensure_quiz(quiz.id, "title")
|
|
|
|
# ── Semantic (vector) search ──────────────────────────────────
|
|
seen_question_ids = set()
|
|
if mode in ("questions", "all"):
|
|
query_emb = embedding_service.generate_embedding(phrase)
|
|
if query_emb:
|
|
# Use f-string for the vector literal — safe because it's a list of floats
|
|
emb_literal = "[" + ",".join(str(float(x)) for x in query_emb) + "]"
|
|
rows = db.execute(sa_text("""
|
|
SELECT q.id, q.quiz_id, q.question_text, q.options,
|
|
q.correct_answer, q.explanation,
|
|
1 - (q.embedding <=> CAST(:vec AS vector)) AS similarity
|
|
FROM questions q
|
|
WHERE q.embedding IS NOT NULL
|
|
ORDER BY q.embedding <=> CAST(:vec AS vector)
|
|
LIMIT 40
|
|
"""), {"vec": emb_literal}).fetchall()
|
|
|
|
for row in rows:
|
|
similarity = float(row.similarity)
|
|
if similarity < 0.30:
|
|
continue
|
|
quiz_id_val = row.quiz_id # DB column still named quiz_id
|
|
if quiz_id_val and _ensure_quiz(quiz_id_val, "questions"):
|
|
seen_question_ids.add(row.id)
|
|
results[quiz_id_val]["matching_questions"].append({
|
|
"id": row.id,
|
|
"question_text": row.question_text,
|
|
"options": row.options,
|
|
"correct_answer": row.correct_answer,
|
|
"explanation": row.explanation,
|
|
"similarity": round(similarity, 3),
|
|
"match_source": "semantic",
|
|
})
|
|
|
|
# ── Keyword (ILIKE) search ────────────────────────────────────
|
|
if mode in ("questions", "all"):
|
|
q_filter = or_(
|
|
QuestionModel.question_text.ilike(f"%{phrase}%"),
|
|
cast(QuestionModel.options, String).ilike(f"%{phrase}%"),
|
|
)
|
|
keyword_rows = (
|
|
db.query(QuestionModel)
|
|
.filter(q_filter)
|
|
.order_by(QuestionModel.source_quiz_id, QuestionModel.id)
|
|
.limit(200)
|
|
.all()
|
|
)
|
|
for question in keyword_rows:
|
|
src_qid = question.source_quiz_id
|
|
if src_qid and _ensure_quiz(src_qid, "questions"):
|
|
if question.id not in seen_question_ids:
|
|
seen_question_ids.add(question.id)
|
|
results[src_qid]["matching_questions"].append({
|
|
"id": question.id,
|
|
"question_text": question.question_text,
|
|
"options": question.options,
|
|
"correct_answer": question.correct_answer,
|
|
"explanation": question.explanation,
|
|
"similarity": None,
|
|
"match_source": "keyword",
|
|
})
|
|
|
|
# Sort each quiz's questions: semantic first (by similarity desc), then keyword
|
|
for r in results.values():
|
|
r["matching_questions"].sort(
|
|
key=lambda x: (0 if x.get("match_source") == "semantic" else 1,
|
|
-(x.get("similarity") or 0))
|
|
)
|
|
|
|
# Sort results: title matches first, then by number of semantic hits desc
|
|
sorted_results = sorted(
|
|
results.values(),
|
|
key=lambda r: (
|
|
0 if r["match_type"] == "title" else 1,
|
|
-sum(1 for q in r["matching_questions"] if q.get("match_source") == "semantic"),
|
|
),
|
|
)
|
|
return sorted_results
|
|
|
|
|
|
@router.get("/", response_model=list[QuizResponse])
|
|
def list_quizzes(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""List quizzes. Moderators see all; regular users only see published."""
|
|
q = db.query(Quiz).filter(Quiz.deleted_at.is_(None))
|
|
if not current_user.is_moderator:
|
|
q = q.filter(Quiz.is_published == 1)
|
|
return q.order_by(Quiz.created_at.desc()).all()
|
|
|
|
|
|
@router.get("/{quiz_id}")
|
|
def get_quiz(
|
|
quiz_id: int,
|
|
study: bool = Query(False),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Get quiz for taking. study=true forces answers/explanations to be included."""
|
|
quiz = db.query(Quiz).filter(Quiz.id == quiz_id, Quiz.deleted_at.is_(None)).first()
|
|
if not quiz:
|
|
raise HTTPException(status_code=404, detail="Quiz not found")
|
|
|
|
if study or quiz.mode == "learning":
|
|
return QuizLearningDetail.model_validate(quiz)
|
|
return QuizDetail.model_validate(quiz)
|
|
|
|
|
|
@router.patch("/{quiz_id}", response_model=QuizResponse)
|
|
def update_quiz(
|
|
quiz_id: int,
|
|
data: QuizUpdate,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_moderator),
|
|
):
|
|
"""Update quiz metadata (title, etc.) — moderator only."""
|
|
quiz = db.query(Quiz).filter(Quiz.id == quiz_id, Quiz.deleted_at.is_(None)).first()
|
|
if not quiz:
|
|
raise HTTPException(status_code=404, detail="Quiz not found")
|
|
|
|
if data.title:
|
|
quiz.title = data.title.strip()
|
|
|
|
db.commit()
|
|
db.refresh(quiz)
|
|
return quiz
|
|
|
|
|
|
@router.post("/{quiz_id}/shuffle", response_model=QuizDetail)
|
|
def shuffle_quiz(
|
|
quiz_id: int,
|
|
shuffle_options: bool = Query(True, description="Also shuffle answer options within each question"),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Return quiz with question order and optionally option order shuffled. Does not modify DB."""
|
|
from app.models.question import Question as QuestionModel
|
|
from app.schemas.quiz import QuestionResponse
|
|
|
|
quiz = db.query(Quiz).filter(Quiz.id == quiz_id).first()
|
|
if not quiz:
|
|
raise HTTPException(status_code=404, detail="Quiz not found")
|
|
|
|
questions = get_quiz_questions(db, quiz_id)
|
|
shuffled_questions = questions.copy()
|
|
random.shuffle(shuffled_questions)
|
|
|
|
result = []
|
|
for q in shuffled_questions:
|
|
options = list(q.options) if q.options else None
|
|
if shuffle_options and options:
|
|
random.shuffle(options)
|
|
result.append(QuestionResponse(
|
|
id=q.id,
|
|
question_text=q.question_text,
|
|
question_type=q.question_type,
|
|
options=options,
|
|
image_path=q.image_path,
|
|
))
|
|
|
|
return {
|
|
"id": quiz.id,
|
|
"section_id": quiz.section_id,
|
|
"user_id": quiz.user_id,
|
|
"title": quiz.title,
|
|
"questions_count": quiz.questions_count,
|
|
"mode": quiz.mode,
|
|
"time_limit_minutes": quiz.time_limit_minutes,
|
|
"created_at": quiz.created_at,
|
|
"questions": result,
|
|
}
|
|
|
|
|
|
@router.get("/{quiz_id}/review", response_model=QuizReview)
|
|
def review_quiz(
|
|
quiz_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Get quiz with answers — only if user has completed an attempt."""
|
|
quiz = db.query(Quiz).filter(Quiz.id == quiz_id).first()
|
|
if not quiz:
|
|
raise HTTPException(status_code=404, detail="Quiz not found")
|
|
|
|
has_attempt = db.query(QuizAttempt).filter(
|
|
QuizAttempt.quiz_id == quiz_id,
|
|
QuizAttempt.user_id == current_user.id,
|
|
QuizAttempt.completed_at.isnot(None),
|
|
).first()
|
|
if not has_attempt and not current_user.is_moderator:
|
|
raise HTTPException(status_code=403, detail="Complete an attempt first to review answers")
|
|
|
|
return quiz
|
|
|
|
|
|
@router.get("/{quiz_id}/questions")
|
|
def get_quiz_questions_for_edit(
|
|
quiz_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_moderator),
|
|
):
|
|
"""Get all questions with answers for editing — moderator/admin only."""
|
|
quiz = db.query(Quiz).filter(Quiz.id == quiz_id).first()
|
|
if not quiz:
|
|
raise HTTPException(status_code=404, detail="Quiz not found")
|
|
questions = get_quiz_questions(db, quiz_id)
|
|
return [
|
|
{
|
|
"id": q.id,
|
|
"question_text": q.question_text,
|
|
"question_type": q.question_type,
|
|
"options": q.options,
|
|
"correct_answer": q.correct_answer,
|
|
"explanation": q.explanation,
|
|
}
|
|
for q in questions
|
|
]
|
|
|
|
|
|
@router.patch("/{quiz_id}/questions/{question_id}")
|
|
def update_question(
|
|
quiz_id: int,
|
|
question_id: int,
|
|
data: dict,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_moderator),
|
|
):
|
|
"""Update a quiz question — change is shared: affects all quizzes using this question."""
|
|
if not question_in_quiz(db, quiz_id, question_id):
|
|
raise HTTPException(status_code=404, detail="Question not found in this quiz")
|
|
question = db.query(QuestionModel).filter(QuestionModel.id == question_id).first()
|
|
if not question:
|
|
raise HTTPException(status_code=404, detail="Question not found")
|
|
|
|
allowed = {"question_text", "options", "correct_answer", "explanation", "question_type"}
|
|
for key, value in data.items():
|
|
if key in allowed:
|
|
setattr(question, key, value)
|
|
|
|
# Validate correct_answer is one of the options
|
|
if question.options and question.correct_answer not in question.options:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"correct_answer must match one of the options exactly. Options: {question.options}"
|
|
)
|
|
|
|
db.commit()
|
|
db.refresh(question)
|
|
return {
|
|
"id": question.id,
|
|
"question_text": question.question_text,
|
|
"question_type": question.question_type,
|
|
"options": question.options,
|
|
"correct_answer": question.correct_answer,
|
|
"explanation": question.explanation,
|
|
}
|
|
|
|
|
|
@router.delete("/{quiz_id}/questions/{question_id}", status_code=204)
|
|
def delete_question(
|
|
quiz_id: int,
|
|
question_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_moderator),
|
|
):
|
|
"""Remove a question from this quiz. If it's shared, only removes the link (question stays in bank).
|
|
If it's only in this quiz, deletes it from the bank too."""
|
|
if not question_in_quiz(db, quiz_id, question_id):
|
|
raise HTTPException(status_code=404, detail="Question not found in this quiz")
|
|
|
|
from app.models.quiz_question_link import QuizQuestionLink
|
|
# Check if used by any OTHER quiz
|
|
other_uses = db.query(QuizQuestionLink).filter(
|
|
QuizQuestionLink.question_id == question_id,
|
|
QuizQuestionLink.quiz_id != quiz_id,
|
|
).count()
|
|
|
|
remove_question_from_quiz(db, quiz_id, question_id)
|
|
|
|
quiz = db.query(Quiz).filter(Quiz.id == quiz_id).first()
|
|
if quiz and quiz.questions_count > 0:
|
|
quiz.questions_count -= 1
|
|
|
|
# Only delete the question record if no other quiz references it
|
|
if other_uses == 0:
|
|
db.query(QuestionModel).filter(QuestionModel.id == question_id).delete()
|
|
|
|
db.commit()
|
|
|
|
|
|
@router.delete("/{quiz_id}", status_code=204)
|
|
def delete_quiz(
|
|
quiz_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_moderator),
|
|
):
|
|
"""Soft-delete quiz — moves to trash. Restore via PATCH /{quiz_id}/restore."""
|
|
quiz = db.query(Quiz).filter(Quiz.id == quiz_id, Quiz.deleted_at.is_(None)).first()
|
|
if not quiz:
|
|
raise HTTPException(status_code=404, detail="Quiz not found")
|
|
quiz.deleted_at = datetime.utcnow()
|
|
db.commit()
|
|
|
|
|
|
@router.patch("/{quiz_id}/publish")
|
|
def set_quiz_published(
|
|
quiz_id: int,
|
|
published: bool = True,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_moderator),
|
|
):
|
|
"""Hide (published=false) or show (published=true) a quiz for regular users."""
|
|
quiz = db.query(Quiz).filter(Quiz.id == quiz_id, Quiz.deleted_at.is_(None)).first()
|
|
if not quiz:
|
|
raise HTTPException(status_code=404, detail="Quiz not found")
|
|
quiz.is_published = 1 if published else 0
|
|
db.commit()
|
|
return {"quiz_id": quiz_id, "is_published": quiz.is_published}
|
|
|
|
|
|
@router.get("/trash", response_model=list[QuizResponse])
|
|
def list_trash(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_moderator),
|
|
):
|
|
"""List soft-deleted quizzes — admin/moderator only."""
|
|
return db.query(Quiz).filter(Quiz.deleted_at.isnot(None)).order_by(Quiz.deleted_at.desc()).all()
|
|
|
|
|
|
@router.patch("/{quiz_id}/restore", response_model=QuizResponse)
|
|
def restore_quiz(
|
|
quiz_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_moderator),
|
|
):
|
|
"""Restore a soft-deleted quiz from trash."""
|
|
quiz = db.query(Quiz).filter(Quiz.id == quiz_id, Quiz.deleted_at.isnot(None)).first()
|
|
if not quiz:
|
|
raise HTTPException(status_code=404, detail="Quiz not found in trash")
|
|
quiz.deleted_at = None
|
|
db.commit()
|
|
db.refresh(quiz)
|
|
return quiz
|
|
|
|
|
|
@router.delete("/{quiz_id}/permanent", status_code=204)
|
|
def permanently_delete_quiz(
|
|
quiz_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(require_moderator),
|
|
):
|
|
"""Permanently delete a quiz that's already in trash. Questions stay in bank."""
|
|
from app.models.quiz_question_link import QuizQuestionLink
|
|
from datetime import datetime as dt
|
|
|
|
quiz = db.query(Quiz).filter(Quiz.id == quiz_id).first()
|
|
if not quiz:
|
|
raise HTTPException(status_code=404, detail="Quiz not found")
|
|
if quiz.deleted_at is None:
|
|
raise HTTPException(status_code=400, detail="Move to trash first before permanently deleting")
|
|
|
|
# Detach exclusive questions to bank
|
|
links = db.query(QuizQuestionLink).filter(QuizQuestionLink.quiz_id == quiz_id).all()
|
|
for link in links:
|
|
other = db.query(QuizQuestionLink).filter(
|
|
QuizQuestionLink.question_id == link.question_id,
|
|
QuizQuestionLink.quiz_id != quiz_id,
|
|
).count()
|
|
if other == 0:
|
|
db.query(QuestionModel).filter(QuestionModel.id == link.question_id).update(
|
|
{QuestionModel.source_quiz_id: None}, synchronize_session=False
|
|
)
|
|
|
|
db.delete(quiz)
|
|
db.commit()
|