from typing import List from repositories import QuizRepository, UserRepository from schemas import QuizGetSchema from schemas.requests import QuizCreateSchema from schemas.response import UserQuizListResponse, RecomendationResponse from exception import DataNotFoundException from mapper import map_quiz_entity_to_schema, quiz_to_recomendation_mapper from exception import ValidationException class QuizService: def __init__(self, quiz_repository=QuizRepository, user_repository=UserRepository): self.quiz_repository = quiz_repository self.user_repostory = user_repository def get_quiz(self, quiz_id) -> QuizGetSchema: data = self.quiz_repository.get_by_id(quiz_id) if data is None: raise DataNotFoundException("Quiz not found") return map_quiz_entity_to_schema(data) def search_quiz( self, keyword: str, page: int = 1, page_size: int = 10 ) -> tuple[list[RecomendationResponse], int]: # if not keyword: # raise ValidationException("Keyword cannot be empty.") quizzes = self.quiz_repository.search_by_title_or_category( keyword, page, page_size ) total = self.quiz_repository.count_by_search(keyword) mapped_quizzes = [] for quiz in quizzes: author = self.user_repostory.get_user_by_id(user_id=quiz.author_id) if author is None: continue mapped_quizzes.append( quiz_to_recomendation_mapper(quiz_entity=quiz, user_entity=author) ) return mapped_quizzes, total def get_user_quiz( self, user_id: str, page: int = 1, page_size: int = 10 ) -> UserQuizListResponse: quizzes = self.quiz_repository.get_by_user_id(user_id, page, page_size) total_user_quiz = self.quiz_repository.count_by_user_id(user_id) return UserQuizListResponse( total=total_user_quiz, quizzes=[map_quiz_entity_to_schema(quiz) for quiz in quizzes], ) def create_quiz(self, quiz_data: QuizCreateSchema): for question in quiz_data.question_listings: if question.type == "option" and ( not question.options or len(question.options) != 4 ): raise ValidationException( "Option type questions must have exactly 4 options." ) return self.quiz_repository.create(quiz_data) def update_quiz(self, quiz_id, quiz_data): return self.quiz_repository.update(quiz_id, quiz_data) def delete_quiz(self, quiz_id): return self.quiz_repository.delete(quiz_id) def get_quiz_recommendation(self, page: int, limit: int): data = self.quiz_repository.get_top_played_quizzes(page=page, limit=limit) if not data: raise DataNotFoundException("Quiz not found") result = [] for quiz in data: author = self.user_repostory.get_user_by_id(user_id=quiz.author_id) result.append( quiz_to_recomendation_mapper(quiz_entity=quiz, user_entity=author) ) return result