6f0470e475
- Setup project structure with FastAPI app - Create SQLAlchemy models for categories, questions, quizzes, and results - Implement API endpoints for all CRUD operations - Set up Alembic migrations for database schema management - Add comprehensive documentation in README.md
82 lines
2.3 KiB
Python
82 lines
2.3 KiB
Python
from sqlalchemy.orm import Session
|
|
from typing import List, Optional
|
|
|
|
from app.models.quiz import Quiz
|
|
from app.models.question import Question
|
|
from app.schemas.quiz import QuizCreate, QuizUpdate
|
|
|
|
|
|
def create_quiz(db: Session, quiz_data: QuizCreate) -> Quiz:
|
|
"""Create a new quiz."""
|
|
# Extract question_ids
|
|
question_ids = quiz_data.question_ids
|
|
quiz_dict = quiz_data.model_dump(exclude={"question_ids"})
|
|
|
|
# Create quiz
|
|
db_quiz = Quiz(**quiz_dict)
|
|
db.add(db_quiz)
|
|
db.commit()
|
|
|
|
# Add questions to quiz
|
|
questions = db.query(Question).filter(Question.id.in_(question_ids)).all()
|
|
db_quiz.questions = questions
|
|
db.commit()
|
|
db.refresh(db_quiz)
|
|
|
|
return db_quiz
|
|
|
|
|
|
def get_quiz(db: Session, quiz_id: int) -> Optional[Quiz]:
|
|
"""Get a quiz by ID."""
|
|
return db.query(Quiz).filter(Quiz.id == quiz_id).first()
|
|
|
|
|
|
def get_quiz_with_questions(db: Session, quiz_id: int) -> Optional[Quiz]:
|
|
"""Get a quiz by ID with questions."""
|
|
return db.query(Quiz).filter(Quiz.id == quiz_id).first()
|
|
|
|
|
|
def get_quizzes(
|
|
db: Session,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
category_id: Optional[int] = None
|
|
) -> List[Quiz]:
|
|
"""Get all quizzes with optional filtering and pagination."""
|
|
query = db.query(Quiz)
|
|
|
|
if category_id:
|
|
query = query.filter(Quiz.category_id == category_id)
|
|
|
|
return query.offset(skip).limit(limit).all()
|
|
|
|
|
|
def update_quiz(db: Session, quiz_id: int, quiz_data: QuizUpdate) -> Optional[Quiz]:
|
|
"""Update an existing quiz."""
|
|
db_quiz = get_quiz(db, quiz_id)
|
|
if not db_quiz:
|
|
return None
|
|
|
|
# Update quiz fields
|
|
update_data = quiz_data.model_dump(exclude={"question_ids"}, exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(db_quiz, key, value)
|
|
|
|
# Update questions if provided
|
|
if quiz_data.question_ids is not None:
|
|
questions = db.query(Question).filter(Question.id.in_(quiz_data.question_ids)).all()
|
|
db_quiz.questions = questions
|
|
|
|
db.commit()
|
|
db.refresh(db_quiz)
|
|
return db_quiz
|
|
|
|
|
|
def delete_quiz(db: Session, quiz_id: int) -> bool:
|
|
"""Delete a quiz."""
|
|
db_quiz = get_quiz(db, quiz_id)
|
|
if db_quiz:
|
|
db.delete(db_quiz)
|
|
db.commit()
|
|
return True
|
|
return False |