98 lines
3.5 KiB
Python
98 lines
3.5 KiB
Python
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
|
|
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(
|
|
RecomendationResponse(
|
|
quiz_id=str(quiz.id),
|
|
author_id=str(author.id),
|
|
author_name=author.name,
|
|
title=quiz.title,
|
|
description=quiz.description,
|
|
date=quiz.date.strftime("%d-%B-%Y"),
|
|
)
|
|
)
|
|
|
|
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(
|
|
RecomendationResponse(
|
|
quiz_id=str(quiz.id),
|
|
author_id=str(author.id),
|
|
author_name=author.name,
|
|
title=quiz.title,
|
|
description=quiz.description,
|
|
date=quiz.date.strftime("%d-%B-%Y"),
|
|
)
|
|
)
|
|
return result
|