fix: entntiy and schema for create quiz

This commit is contained in:
akhdanre 2025-05-03 15:41:29 +07:00
parent fa01256c71
commit 5774850aaa
14 changed files with 163 additions and 97 deletions

View File

@ -33,28 +33,10 @@ class QuizController:
status_code=201,
)
except (ValidationError, ValidationException) as e:
return make_response(e.errors(), status_code=400)
return make_response(message="", status_code=400)
except Exception as e:
return make_error_response(e)
# def update_quiz(self, quiz_id, quiz_data):
# try:
# quiz_obj = QuizUpdateSchema(**quiz_data)
# success = self.quiz_service.update_quiz(
# quiz_id, quiz_obj.dict(exclude_unset=True)
# )
# if not success:
# return jsonify({"error": "Quiz not found or update failed"}), 400
# return jsonify({"message": "Quiz updated"}), 200
# except ValidationError as e:
# return jsonify({"error": "Validation error", "detail": e.errors()}), 400
# def delete_quiz(self, quiz_id):
# success = self.quiz_service.delete_quiz(quiz_id)
# if not success:
# return jsonify({"error": "Quiz not found"}), 400
# return jsonify({"message": "Quiz deleted"}), 200
def quiz_recomendation(self):
try:
result = self.quiz_service.get_quiz_recommendation()
@ -114,6 +96,8 @@ class QuizController:
return make_response(
message="success retrieve recommendation quiz", data=result
)
except DataNotFoundException as e:
return make_response(message=e.message, status_code=e.status_code)
except ValueError as e:
return make_response(message=str(e), data=None, status_code=400)
except ValidationError as e:

View File

@ -1,4 +1,9 @@
from .response_helper import make_response, make_error_response
from .datetime_util import DatetimeUtil
__all__ = ["make_response", "make_error_response"]
__all__ = [
"make_response",
"make_error_response",
"DatetimeUtil",
]

View File

@ -0,0 +1,35 @@
from datetime import datetime
from zoneinfo import ZoneInfo
class DatetimeUtil:
@staticmethod
def now():
"""Waktu UTC (timezone-aware)"""
return datetime.now(tz=ZoneInfo("UTC"))
@staticmethod
def now_iso():
"""Waktu UTC dalam format ISO 8601 string"""
return datetime.now(tz=ZoneInfo("UTC")).isoformat()
@staticmethod
def now_jakarta():
"""Waktu sekarang di zona Asia/Jakarta (WIB)"""
return datetime.now(tz=ZoneInfo("Asia/Jakarta"))
@staticmethod
def to_string(dt: datetime, fmt: str = "%Y-%m-%d %H:%M:%S") -> str:
"""Convert UTC datetime to Asia/Jakarta time and format as string"""
if dt.tzinfo is None:
dt = dt.replace(tzinfo=ZoneInfo("UTC"))
jakarta_time = dt.astimezone(ZoneInfo("Asia/Jakarta"))
return jakarta_time.strftime(fmt)
@staticmethod
def from_string(
date_str: str, fmt: str = "%Y-%m-%d %H:%M:%S", tz: str = "UTC"
) -> datetime:
"""Convert string ke datetime dengan timezone"""
dt = datetime.strptime(date_str, fmt)
return dt.replace(tzinfo=ZoneInfo(tz))

View File

@ -1,9 +1,8 @@
from .user_mapper import UserMapper
from .quiz_mapper import map_quiz_entity_to_schema, quiz_to_recomendation_mapper
from .quiz_mapper import QuizMapper
__all__ = [
"UserMapper",
"map_quiz_entity_to_schema",
"quiz_to_recomendation_mapper",
"QuizMapper",
]

View File

@ -1,46 +1,85 @@
from datetime import datetime
from helpers import DatetimeUtil
from models import QuizEntity, QuestionItemEntity, UserEntity
from schemas import QuizGetSchema, QuestionItemSchema
from schemas.response import RecomendationResponse
from schemas.requests import QuizCreateSchema
def map_question_entity_to_schema(entity: QuestionItemEntity) -> QuestionItemSchema:
return QuestionItemSchema(
index=entity.index,
question=entity.question,
target_answer=entity.target_answer,
duration=entity.duration,
type=entity.type,
options=entity.options,
)
class QuizMapper:
@staticmethod
def map_question_entity_to_schema(entity: QuestionItemEntity) -> QuestionItemSchema:
return QuestionItemSchema(
index=entity.index,
question=entity.question,
target_answer=entity.target_answer,
duration=entity.duration,
type=entity.type,
options=entity.options,
)
def map_quiz_entity_to_schema(entity: QuizEntity) -> QuizGetSchema:
return QuizGetSchema(
id=str(entity.id),
author_id=entity.author_id,
title=entity.title,
description=entity.description,
is_public=entity.is_public,
date=entity.date.strftime("%Y-%m-%d %H:%M:%S") if entity.date else None,
total_quiz=entity.total_quiz or 0,
limit_duration=entity.limit_duration or 0,
question_listings=[
map_question_entity_to_schema(q) for q in entity.question_listings or []
],
)
@staticmethod
def map_question_schema_to_entity(schema: QuestionItemSchema) -> QuestionItemEntity:
return QuestionItemEntity(
index=schema.index,
question=schema.question,
target_answer=schema.target_answer,
duration=schema.duration,
type=schema.type,
options=schema.options,
)
@staticmethod
def map_quiz_entity_to_schema(entity: QuizEntity) -> QuizGetSchema:
return QuizGetSchema(
id=str(entity.id),
author_id=entity.author_id,
title=entity.title,
description=entity.description,
is_public=entity.is_public,
date=DatetimeUtil.to_string(entity.date, "%d-%m-%Y"),
time=DatetimeUtil.to_string(entity.date, "%H:%M:%S"),
total_quiz=entity.total_quiz or 0,
limit_duration=entity.limit_duration or 0,
question_listings=[
QuizMapper.map_question_entity_to_schema(q)
for q in entity.question_listings or []
],
)
def quiz_to_recomendation_mapper(
quiz_entity: QuizEntity,
user_entity: UserEntity,
) -> RecomendationResponse:
return RecomendationResponse(
quiz_id=str(quiz_entity.id),
author_id=str(user_entity.id),
author_name=user_entity.name,
title=quiz_entity.title,
description=quiz_entity.description,
date=quiz_entity.date.strftime("%d-%B-%Y"),
duration=quiz_entity.limit_duration,
total_quiz=quiz_entity.total_quiz,
)
@staticmethod
def map_quiz_schema_to_entity(
schema: QuizCreateSchema,
datetime: datetime,
total_duration: int,
) -> QuizEntity:
return QuizEntity(
author_id=schema.author_id,
title=schema.title,
description=schema.description,
is_public=schema.is_public,
date=datetime,
total_quiz=len(schema.question_listings),
limit_duration=total_duration,
question_listings=[
QuizMapper.map_question_schema_to_entity(q)
for q in schema.question_listings or []
],
)
@staticmethod
def quiz_to_recomendation_mapper(
quiz_entity: QuizEntity,
user_entity: UserEntity,
) -> RecomendationResponse:
return RecomendationResponse(
quiz_id=str(quiz_entity.id),
author_id=str(user_entity.id),
author_name=user_entity.name,
title=quiz_entity.title,
description=quiz_entity.description,
date=quiz_entity.date.strftime("%d-%B-%Y") if quiz_entity.date else None,
duration=quiz_entity.limit_duration,
total_quiz=quiz_entity.total_quiz,
)

View File

@ -1,11 +1,11 @@
from typing import Optional, List
from typing import Optional, List, Union
from pydantic import BaseModel
class QuestionItemEntity(BaseModel):
index: int
question: str
target_answer: str
target_answer: Union[str, bool, int]
duration: int
type: str
options: Optional[List[str]] = None

View File

@ -6,15 +6,15 @@ from .question_item_entity import QuestionItemEntity
class QuizEntity(BaseModel):
id: Optional[PyObjectId] = Field(alias="_id")
id: Optional[PyObjectId] = Field(default=None, alias="_id")
author_id: Optional[str] = None
title: str
description: Optional[str] = None
# subject: str
is_public: bool = False
date: Optional[datetime] = None
total_quiz: Optional[int] = 0
limit_duration: Optional[int] = 0 # in minute
date: datetime
total_quiz: int = 0
limit_duration: Optional[int] = 0 # in
total_user_playing: int = 0
question_listings: Optional[list[QuestionItemEntity]] = []

View File

@ -4,6 +4,7 @@ from models import QuizEntity
from pymongo.database import Database
from pymongo.collection import Collection
from datetime import datetime
from helpers import DatetimeUtil
class QuizRepository:
@ -11,7 +12,7 @@ class QuizRepository:
self.collection: Collection = db.quiz
def create(self, quiz: QuizEntity) -> str:
quiz_dict = quiz.dict(by_alias=True, exclude_none=True)
quiz_dict = quiz.model_dump(by_alias=True, exclude_none=True)
result = self.collection.insert_one(quiz_dict)
return str(result.inserted_id)
@ -102,17 +103,7 @@ class QuizRepository:
self.collection.find({"author_id": user_id}).skip(skip).limit(page_size)
)
quiz_list = []
for doc in cursor:
if "date" in doc and isinstance(doc["date"], str):
try:
doc["date"] = datetime.strptime(doc["date"], "%d-%m-%Y")
except ValueError:
doc["date"] = None
quiz_list.append(QuizEntity(**doc))
return quiz_list
return [QuizEntity(**data) for data in cursor]
def get_all(self, skip: int = 0, limit: int = 10) -> List[QuizEntity]:
cursor = self.collection.find().skip(skip).limit(limit)

View File

@ -8,8 +8,5 @@ class QuizCreateSchema(BaseModel):
title: str
description: Optional[str] = None
is_public: bool = False
date: Optional[str] = None
total_quiz: Optional[int] = 0
limit_duration: Optional[int] = 0
author_id: Optional[str] = None
question_listings: Optional[List[QuestionItemSchema]] = []

View File

@ -1,11 +1,11 @@
from typing import List, Optional
from typing import List, Optional, Union
from pydantic import BaseModel
class QuestionItemSchema(BaseModel):
index: int
question: str
target_answer: str
target_answer: Union[str, bool, int]
duration: int
type: str
options: Optional[List[str]] = None

View File

@ -1,11 +1,11 @@
from typing import List, Optional
from typing import List, Optional, Union
from pydantic import BaseModel
class QuestionItemSchema(BaseModel):
index: int
question: str
target_answer: str
target_answer: Union[str | int | bool]
duration: int
type: str
options: Optional[List[str]] = None

View File

@ -10,7 +10,8 @@ class QuizGetSchema(BaseModel):
title: str
description: Optional[str] = None
is_public: bool = False
date: Optional[str] = None
date: str
time: str
total_quiz: int = 0
limit_duration: int = 0
question_listings: List[QuestionItemSchema] = []

View File

@ -9,5 +9,5 @@ class GetSubjectResponse(BaseModel):
description: Optional[str]
class Config:
orm_mode = True
allow_population_by_field_name = True
from_attributes = True
populate_by_name = True

View File

@ -1,11 +1,12 @@
from typing import List
from models import QuizEntity
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 mapper import QuizMapper
from exception import ValidationException
from helpers import DatetimeUtil
class QuizService:
@ -17,13 +18,11 @@ class QuizService:
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)
return QuizMapper.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
@ -35,7 +34,10 @@ class QuizService:
if author is None:
continue
mapped_quizzes.append(
quiz_to_recomendation_mapper(quiz_entity=quiz, user_entity=author)
QuizMapper.quiz_to_recomendation_mapper(
quiz_entity=quiz,
user_entity=author,
)
)
return mapped_quizzes, total
@ -47,10 +49,11 @@ class QuizService:
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],
quizzes=[QuizMapper.map_quiz_entity_to_schema(quiz) for quiz in quizzes],
)
def create_quiz(self, quiz_data: QuizCreateSchema):
total_time = 0
for question in quiz_data.question_listings:
if question.type == "option" and (
not question.options or len(question.options) != 4
@ -58,8 +61,17 @@ class QuizService:
raise ValidationException(
"Option type questions must have exactly 4 options."
)
total_time += question.duration
return self.quiz_repository.create(quiz_data)
datetime_now = DatetimeUtil.now_iso()
data = QuizMapper.map_quiz_schema_to_entity(
schema=quiz_data,
datetime=datetime_now,
total_duration=total_time,
)
return self.quiz_repository.create(data)
def update_quiz(self, quiz_id, quiz_data):
return self.quiz_repository.update(quiz_id, quiz_data)
@ -77,6 +89,9 @@ class QuizService:
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)
QuizMapper.quiz_to_recomendation_mapper(
quiz_entity=quiz,
user_entity=author,
)
)
return result